import { Inject, Injectable, Logger, forwardRef } from "@nestjs/common";
import { ConfigService } from "@nestjs/config";
import { Surreal } from "surrealdb.node";
import { KafkaService, SurrealDBService } from "..";

@Injectable()
export class SurrealAsSinkService {
  constructor(
    @Inject(forwardRef(() => ConfigService))
    private configService: ConfigService,
    @Inject(forwardRef(() => KafkaService))
    private kafkaService: KafkaService,
    @Inject(forwardRef(() => SurrealDBService))
    private dbService: SurrealDBService
  ) {}

  async onModuleInit() {
    const topicToTableMap = this.configService.get(
      "surrealAsSink.topicToTableMap"
    ) as Record<string, string>;
    if (topicToTableMap) {
      const groupId = this.configService.get("surrealAsSink.groupId");
      Object.entries(topicToTableMap).forEach(([topic, table]) => {
        this.kafkaService
          .getReadStream(topic, groupId)
          .subscribe(({ key, value }) => {
            if (value) {
              this.dbService.db.query(
                `INSERT INTO ${table} (${Object.keys(value).map(
                  (key) => `\`${key}\``
                )}) VALUES (${Object.values(value)
                  .map((value) => JSON.stringify(value))
                  .join(",")})`
              );
            } else {
              this.dbService.db.query(
                `DELETE FROM ${table} WHERE meta::id(id) = '${key.id}' `
              );
            }
          });
      });
    } else {
      Logger.warn(
        "topicToTableMap config not set (surrealAsSink.topicToTableMap)",
        "surreal-as-sink service"
      );
    }
  }
}