#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <signal.h>
#include <inttypes.h>
#include <rdkafka.h>
#include <unistd.h>
#include <time.h>

static volatile sig_atomic_t run = 1;
static void stop(int sig) { run = 0; }

static void get_time_str(char *timebuf, size_t size) {
    time_t now = time(NULL);
    struct tm *t = localtime(&now);
    strftime(timebuf, size, "%Y-%m-%d %H:%M:%S", t);
}

static void log_cb(const rd_kafka_t *rk, int level,
                   const char *fac, const char *buf) {
    (void)rk;
    struct timeval tv;
    char timebuf[64];

    gettimeofday(&tv, NULL);
    strftime(timebuf, sizeof(timebuf),
             "%Y-%m-%d %H:%M:%S", localtime(&tv.tv_sec));

    fprintf(stderr, "[%s.%03ld] LOG_CB level=%d fac=%s: %s\n",
            timebuf, tv.tv_usec / 1000, level, fac, buf);
}

static void error_cb(rd_kafka_t *rk, int err,
                     const char *reason, void *opaque) {
    (void)rk; (void)opaque;
    char timebuf[64];
    get_time_str(timebuf, sizeof(timebuf));
    fprintf(stderr, "[%s] ERROR_CB: %s: %s\n",
            timebuf, rd_kafka_err2str(err), reason);
}

static void rebalance_cb(rd_kafka_t *rk,
                         rd_kafka_resp_err_t err,
                         rd_kafka_topic_partition_list_t *partitions,
                         void *opaque) {
    (void)opaque;
    char timebuf[64];
    get_time_str(timebuf, sizeof(timebuf));

    fprintf(stderr, "[%s] REBALANCE_CB: %s, partitions=%d\n",
            timebuf, rd_kafka_err2str(err),
            partitions ? partitions->cnt : -1);

    if (err == RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS)
        rd_kafka_assign(rk, partitions);
    else
        rd_kafka_assign(rk, NULL);
}

int main() {
    signal(SIGINT, stop);

    char *brokers = getenv("BOOTSTRAP_SERVERS");
    char *topic   = getenv("TOPIC");
    char *group   = getenv("GROUP_ID");

    if (!brokers || !topic || !group) {
        fprintf(stderr, "set BOOTSTRAP_SERVERS, TOPIC, GROUP_ID\n");
        return 1;
    }

    rd_kafka_conf_t *conf = rd_kafka_conf_new();
    rd_kafka_conf_set(conf, "bootstrap.servers", brokers, NULL, 0);
    rd_kafka_conf_set(conf, "group.id", group, NULL, 0);
    rd_kafka_conf_set(conf, "auto.offset.reset", "earliest", NULL, 0);
    rd_kafka_conf_set(conf, "metadata.recovery.strategy",
                      "rebootstrap", NULL, 0);
    rd_kafka_conf_set(conf, "debug", "all", NULL, 0);
    rd_kafka_conf_set(conf, "log_level", "7", NULL, 0);

    rd_kafka_conf_set_log_cb(conf, log_cb);
    rd_kafka_conf_set_error_cb(conf, error_cb);
    rd_kafka_conf_set_rebalance_cb(conf, rebalance_cb);

    rd_kafka_t *rk =
        rd_kafka_new(RD_KAFKA_CONSUMER, conf, NULL, 0);
    rd_kafka_poll_set_consumer(rk);

    rd_kafka_topic_partition_list_t *subs =
        rd_kafka_topic_partition_list_new(1);
    rd_kafka_topic_partition_list_add(subs, topic, -1);
    rd_kafka_subscribe(rk, subs);

    while (run) {
        rd_kafka_message_t *msg =
            rd_kafka_consumer_poll(rk, 1000);
        if (!msg) {
            fprintf(stderr, "poll timeout\n");
            continue;
        }
        if (msg->err)
            fprintf(stderr, "msg error: %s\n",
                    rd_kafka_message_errstr(msg));
        rd_kafka_message_destroy(msg);
        sleep(1);
    }

    rd_kafka_consumer_close(rk);
    rd_kafka_destroy(rk);
    return 0;
}