/*
 * librdkafka - The Apache Kafka C/C++ library
 *
 * Copyright (c) 2015 Magnus Edenhill
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 * 1. Redistributions of source code must retain the above copyright notice,
 *    this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright notice,
 *    this list of conditions and the following disclaimer in the documentation
 *    and/or other materials provided with the distribution.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 */
#include "rdkafka_int.h"
#include "rdkafka_assignor.h"

#include <ctype.h>

/**
 * Clear out and free any memory used by the member, but not the rkgm itself.
 */
void rd_kafka_group_member_clear (rd_kafka_group_member_t *rkgm) {
        if (rkgm->rkgm_subscription)
                rd_kafka_topic_partition_list_destroy(rkgm->rkgm_subscription);

        if (rkgm->rkgm_assignment)
                rd_kafka_topic_partition_list_destroy(rkgm->rkgm_assignment);

        rd_list_destroy(&rkgm->rkgm_eligible);

        if (rkgm->rkgm_member_id)
                rd_kafkap_str_destroy(rkgm->rkgm_member_id);

        if (rkgm->rkgm_userdata)
                rd_kafkap_bytes_destroy(rkgm->rkgm_userdata);

        if (rkgm->rkgm_member_metadata)
                rd_kafkap_bytes_destroy(rkgm->rkgm_member_metadata);

        memset(rkgm, 0, sizeof(*rkgm));
}


/**
 * Member id string comparator (takes rd_kafka_group_member_t *)
 */
int rd_kafka_group_member_cmp (const void *_a, const void *_b) {
        const rd_kafka_group_member_t *a =
                (const rd_kafka_group_member_t *)_a;
        const rd_kafka_group_member_t *b =
                (const rd_kafka_group_member_t *)_b;

        return rd_kafkap_str_cmp(a->rkgm_member_id, b->rkgm_member_id);
}


/**
 * Returns true if member subscribes to topic, else false.
 */
int
rd_kafka_group_member_find_subscription (rd_kafka_t *rk,
					 const rd_kafka_group_member_t *rkgm,
					 const char *topic) {
	int i;

	/* Match against member's subscription. */
        for (i = 0 ; i < rkgm->rkgm_subscription->cnt ; i++) {
                const rd_kafka_topic_partition_t *rktpar =
                        &rkgm->rkgm_subscription->elems[i];

		if (rd_kafka_topic_partition_match(rk, rkgm, rktpar,
						   topic, NULL))
			return 1;
	}

	return 0;
}



static rd_kafkap_bytes_t *
rd_kafka_consumer_protocol_member_metadata_new (
	const rd_list_t *topics,
        const void *userdata, size_t userdata_size) {
        rd_kafka_buf_t *rkbuf;
        rd_kafkap_bytes_t *kbytes;
        int i;
	int topic_cnt = rd_list_cnt(topics);
	const rd_kafka_topic_info_t *tinfo;
        size_t len;

        /*
         * MemberMetadata => Version Subscription AssignmentStrategies
         *   Version      => int16
         *   Subscription => Topics UserData
         *     Topics     => [String]
         *     UserData     => Bytes
         */

        rkbuf = rd_kafka_buf_new(1, 100 + (topic_cnt * 100) + userdata_size);

        rd_kafka_buf_write_i16(rkbuf, 0);
        rd_kafka_buf_write_i32(rkbuf, topic_cnt);
	RD_LIST_FOREACH(tinfo, topics, i)
                rd_kafka_buf_write_str(rkbuf, tinfo->topic, -1);
	if (userdata)
		rd_kafka_buf_write_bytes(rkbuf, userdata, userdata_size);
	else /* Kafka 0.9.0.0 cant parse NULL bytes, so we provide empty. */
		rd_kafka_buf_write_bytes(rkbuf, "", 0);

        /* Get binary buffer and allocate a new Kafka Bytes with a copy. */
        rd_slice_init_full(&rkbuf->rkbuf_reader, &rkbuf->rkbuf_buf);
        len = rd_slice_remains(&rkbuf->rkbuf_reader);
        kbytes = rd_kafkap_bytes_new(NULL, (int32_t)len);
        rd_slice_read(&rkbuf->rkbuf_reader, (void *)kbytes->data, len);
        rd_kafka_buf_destroy(rkbuf);

        return kbytes;

}




rd_kafkap_bytes_t *
rd_kafka_assignor_get_metadata (rd_kafka_assignor_t *rkas,
				const rd_list_t *topics) {
        return rd_kafka_consumer_protocol_member_metadata_new(
                topics, rkas->rkas_userdata,
                rkas->rkas_userdata_size);
}





/**
 * Returns 1 if all subscriptions are satifised for this member, else 0.
 */
static int rd_kafka_member_subscription_match (
        rd_kafka_cgrp_t *rkcg,
        rd_kafka_group_member_t *rkgm,
        const rd_kafka_metadata_topic_t *topic_metadata,
        rd_kafka_assignor_topic_t *eligible_topic) {
        int i;
        int has_regex = 0;
        int matched = 0;

        /* Match against member's subscription. */
        for (i = 0 ; i < rkgm->rkgm_subscription->cnt ; i++) {
                const rd_kafka_topic_partition_t *rktpar =
                        &rkgm->rkgm_subscription->elems[i];
		int matched_by_regex = 0;

		if (rd_kafka_topic_partition_match(rkcg->rkcg_rk, rkgm, rktpar,
						   topic_metadata->topic,
						   &matched_by_regex)) {
			rd_list_add(&rkgm->rkgm_eligible,
				    (void *)topic_metadata);
			matched++;
			has_regex += matched_by_regex;
		}
	}

        if (matched)
                rd_list_add(&eligible_topic->members, rkgm);

        if (!has_regex &&
            rd_list_cnt(&rkgm->rkgm_eligible) == rkgm->rkgm_subscription->cnt)
                return 1; /* All subscriptions matched */
        else
                return 0;
}


static void
rd_kafka_assignor_topic_destroy (rd_kafka_assignor_topic_t *at) {
        rd_list_destroy(&at->members);
        rd_free(at);
}

int rd_kafka_assignor_topic_cmp (const void *_a, const void *_b) {
        const rd_kafka_assignor_topic_t *a =
                *(const rd_kafka_assignor_topic_t * const *)_a;
        const rd_kafka_assignor_topic_t *b =
                *(const rd_kafka_assignor_topic_t * const *)_b;

        return !strcmp(a->metadata->topic, b->metadata->topic);
}

/**
 * Maps the available topics to the group members' subscriptions
 * and updates the `member` map with the proper list of eligible topics,
 * the latter are returned in `eligible_topics`.
 */
static void
rd_kafka_member_subscriptions_map (rd_kafka_cgrp_t *rkcg,
                                   rd_list_t *eligible_topics,
                                   const rd_kafka_metadata_t *metadata,
                                   rd_kafka_group_member_t *members,
                                   int member_cnt) {
        int ti;
        rd_kafka_assignor_topic_t *eligible_topic = NULL;

        rd_list_init(eligible_topics, RD_MIN(metadata->topic_cnt, 10),
                     (void *)rd_kafka_assignor_topic_destroy);

        /* For each topic in the cluster, scan through the member list
         * to find matching subscriptions. */
        for (ti = 0 ; ti < metadata->topic_cnt ; ti++) {
                int complete_cnt = 0;
                int i;

                /* Ignore topics in blacklist */
                if (rkcg->rkcg_rk->rk_conf.topic_blacklist &&
		    rd_kafka_pattern_match(rkcg->rkcg_rk->rk_conf.
                                           topic_blacklist,
                                           metadata->topics[ti].topic)) {
                        rd_kafka_dbg(rkcg->rkcg_rk, TOPIC, "BLACKLIST",
                                   "Assignor ignoring blacklisted "
                                     "topic \"%s\"",
                                     metadata->topics[ti].topic);
                        continue;
                }

                if (!eligible_topic)
                        eligible_topic = rd_calloc(1, sizeof(*eligible_topic));

                rd_list_init(&eligible_topic->members, member_cnt, NULL);

                /* For each member: scan through its topic subscription */
                for (i = 0 ; i < member_cnt ; i++) {
                        /* Match topic against existing metadata,
                           incl regex matching. */
                        if (rd_kafka_member_subscription_match(
                                    rkcg, &members[i], &metadata->topics[ti],
                                    eligible_topic))
                                complete_cnt++;
                }

                if (rd_list_empty(&eligible_topic->members)) {
                        rd_list_destroy(&eligible_topic->members);
                        continue;
                }

                eligible_topic->metadata = &metadata->topics[ti];
                rd_list_add(eligible_topics, eligible_topic);
                eligible_topic = NULL;

                if (complete_cnt == (int)member_cnt)
                        break;
        }

        if (eligible_topic)
                rd_free(eligible_topic);
}


rd_kafka_resp_err_t
rd_kafka_assignor_run (rd_kafka_cgrp_t *rkcg,
                       const char *protocol_name,
                       rd_kafka_metadata_t *metadata,
                       rd_kafka_group_member_t *members,
                       int member_cnt,
                       char *errstr, size_t errstr_size) {
        rd_kafka_resp_err_t err;
        rd_kafka_assignor_t *rkas;
        rd_ts_t ts_start = rd_clock();
        int i;
        rd_list_t eligible_topics;
        int j;

	if (!(rkas = rd_kafka_assignor_find(rkcg->rkcg_rk, protocol_name)) ||
	    !rkas->rkas_enabled) {
		rd_snprintf(errstr, errstr_size,
			    "Unsupported assignor \"%s\"", protocol_name);
		return RD_KAFKA_RESP_ERR__UNKNOWN_PROTOCOL;
	}


        /* Map available topics to subscribing members */
        rd_kafka_member_subscriptions_map(rkcg, &eligible_topics, metadata,
                                          members, member_cnt);


        if (rkcg->rkcg_rk->rk_conf.debug & RD_KAFKA_DBG_CGRP) {
                rd_kafka_dbg(rkcg->rkcg_rk, CGRP, "ASSIGN",
                             "Group \"%s\" running %s assignment for "
                             "%d member(s):",
                             rkcg->rkcg_group_id->str, protocol_name,
                             member_cnt);

                for (i = 0 ; i < member_cnt ; i++) {
                        const rd_kafka_group_member_t *member = &members[i];

                        rd_kafka_dbg(rkcg->rkcg_rk, CGRP, "ASSIGN",
                                     " Member \"%.*s\"%s with "
                                     "%d subscription(s):",
                                     RD_KAFKAP_STR_PR(member->rkgm_member_id),
                                     !rd_kafkap_str_cmp(member->rkgm_member_id,
                                                        rkcg->rkcg_member_id) ?
                                     " (me)":"",
                                     member->rkgm_subscription->cnt);
                        for (j = 0 ; j < member->rkgm_subscription->cnt ; j++) {
                                const rd_kafka_topic_partition_t *p =
                                        &member->rkgm_subscription->elems[j];
                                rd_kafka_dbg(rkcg->rkcg_rk, CGRP, "ASSIGN",
                                             "  %s [%"PRId32"]",
                                             p->topic, p->partition);
                        }
                }


        }

        /* Call assignors assign callback */
        err = rkas->rkas_assign_cb(rkcg->rkcg_rk,
                                    rkcg->rkcg_member_id->str,
                                    protocol_name, metadata,
                                    members, member_cnt,
                                    (rd_kafka_assignor_topic_t **)
                                    eligible_topics.rl_elems,
                                    eligible_topics.rl_cnt,
                                    errstr, sizeof(errstr),
                                    rkas->rkas_opaque);

        if (err) {
                rd_kafka_dbg(rkcg->rkcg_rk, CGRP, "ASSIGN",
                             "Group \"%s\" %s assignment failed "
                             "for %d member(s): %s",
                             rkcg->rkcg_group_id->str, protocol_name,
                             (int)member_cnt, errstr);
        } else if (rkcg->rkcg_rk->rk_conf.debug & RD_KAFKA_DBG_CGRP) {
                rd_kafka_dbg(rkcg->rkcg_rk, CGRP, "ASSIGN",
                             "Group \"%s\" %s assignment for %d member(s) "
                             "finished in %.3fms:",
                             rkcg->rkcg_group_id->str, protocol_name,
                             (int)member_cnt,
                             (float)(rd_clock() - ts_start)/1000.0f);
                for (i = 0 ; i < member_cnt ; i++) {
                        const rd_kafka_group_member_t *member = &members[i];

                        rd_kafka_dbg(rkcg->rkcg_rk, CGRP, "ASSIGN",
                                     " Member \"%.*s\"%s assigned "
                                     "%d partition(s):",
                                     RD_KAFKAP_STR_PR(member->rkgm_member_id),
                                     !rd_kafkap_str_cmp(member->rkgm_member_id,
                                                        rkcg->rkcg_member_id) ?
                                     " (me)":"",
                                     member->rkgm_assignment->cnt);
                        for (j = 0 ; j < member->rkgm_assignment->cnt ; j++) {
                                const rd_kafka_topic_partition_t *p =
                                        &member->rkgm_assignment->elems[j];
                                rd_kafka_dbg(rkcg->rkcg_rk, CGRP, "ASSIGN",
                                             "  %s [%"PRId32"]",
                                             p->topic, p->partition);
                        }
                }
        }

        rd_list_destroy(&eligible_topics);

        return err;
}


/**
 * Assignor protocol string comparator
 */
static int rd_kafka_assignor_cmp_str (const void *_a, const void *_b) {
        const char *a = _a;
        const rd_kafka_assignor_t *b = _b;

        return rd_kafkap_str_cmp_str2(a, b->rkas_protocol_name);
}

/**
 * Find assignor by protocol name.
 *
 * Locality: any
 * Locks: none
 */
rd_kafka_assignor_t *
rd_kafka_assignor_find (rd_kafka_t *rk, const char *protocol) {
        return (rd_kafka_assignor_t *)
                rd_list_find(&rk->rk_conf.partition_assignors, protocol,
                             rd_kafka_assignor_cmp_str);
}


/**
 * Destroys an assignor (but does not unlink).
 */
static void rd_kafka_assignor_destroy (rd_kafka_assignor_t *rkas) {
        rd_kafkap_str_destroy(rkas->rkas_protocol_type);
        rd_kafkap_str_destroy(rkas->rkas_protocol_name);
        rd_free(rkas);
}



/**
 * Add an assignor, overwriting any previous one with the same protocol_name.
 */
static rd_kafka_resp_err_t
rd_kafka_assignor_add (rd_kafka_t *rk,
		       rd_kafka_assignor_t **rkasp,
                       const char *protocol_type,
                       const char *protocol_name,
                       rd_kafka_resp_err_t (*assign_cb) (
                               rd_kafka_t *rk,
                               const char *member_id,
                               const char *protocol_name,
                               const rd_kafka_metadata_t *metadata,
                               rd_kafka_group_member_t *members,
                               size_t member_cnt,
                               rd_kafka_assignor_topic_t **eligible_topics,
                               size_t eligible_topic_cnt,
                               char *errstr, size_t errstr_size, void *opaque),
                       void *opaque) {
        rd_kafka_assignor_t *rkas;

	if (rkasp)
		*rkasp = NULL;

        if (rd_kafkap_str_cmp_str(rk->rk_conf.group_protocol_type,
                                  protocol_type))
                return RD_KAFKA_RESP_ERR__UNKNOWN_PROTOCOL;

        /* Dont overwrite application assignors */
        if ((rkas = rd_kafka_assignor_find(rk, protocol_name))) {
		if (rkasp)
			*rkasp = rkas;
		return RD_KAFKA_RESP_ERR__CONFLICT;
	}

        rkas = rd_calloc(1, sizeof(*rkas));

        rkas->rkas_protocol_name    = rd_kafkap_str_new(protocol_name, -1);
        rkas->rkas_protocol_type    = rd_kafkap_str_new(protocol_type, -1);
        rkas->rkas_assign_cb        = assign_cb;
        rkas->rkas_get_metadata_cb  = rd_kafka_assignor_get_metadata;
        rkas->rkas_opaque = opaque;

        rd_list_add(&rk->rk_conf.partition_assignors, rkas);

	if (rkasp)
		*rkasp = rkas;

        return RD_KAFKA_RESP_ERR_NO_ERROR;
}


/* Right trim string of whitespaces */
static void rtrim (char *s) {
	char *e = s + strlen(s);

	if (e == s)
		return;

	while (e >= s && isspace(*e))
		e--;

	*e = '\0';
}


/**
 * Initialize assignor list based on configuration.
 */
int rd_kafka_assignors_init (rd_kafka_t *rk, char *errstr, size_t errstr_size) {
	char *wanted;
	char *s;

        rd_list_init(&rk->rk_conf.partition_assignors, 2,
                     (void *)rd_kafka_assignor_destroy);

	rd_strdupa(&wanted, rk->rk_conf.partition_assignment_strategy);

	s = wanted;
	while (*s) {
		rd_kafka_assignor_t *rkas = NULL;
		char *t;

		/* Left trim */
		while (*s == ' ' || *s == ',')
			s++;

		if ((t = strchr(s, ','))) {
			*t = '\0';
			t++;
		} else {
			t = s + strlen(s);
		}

		/* Right trim */
		rtrim(s);

		/* Match builtin consumer assignors */
		if (!strcmp(s, "range"))
			rd_kafka_assignor_add(
				rk, &rkas, "consumer", "range",
				rd_kafka_range_assignor_assign_cb,
				NULL);
		else if (!strcmp(s, "roundrobin"))
			rd_kafka_assignor_add(
				rk, &rkas, "consumer", "roundrobin",
				rd_kafka_roundrobin_assignor_assign_cb,
				NULL);
		else {
			rd_snprintf(errstr, errstr_size,
				    "Unsupported partition.assignment.strategy:"
				    " %s", s);
			return -1;
		}

		if (rkas) {
			if (!rkas->rkas_enabled) {
				rkas->rkas_enabled = 1;
				rk->rk_conf.enabled_assignor_cnt++;
			}
		}

		s = t;
	}

	return 0;
}



/**
 * Free assignors
 */
void rd_kafka_assignors_term (rd_kafka_t *rk) {
        rd_list_destroy(&rk->rk_conf.partition_assignors);
}