/**
 * Incremental embedding worker.
 *
 * Listens to the `message_inserted` Postgres NOTIFY channel. When a new
 * message arrives, it embeds the body via model-boss and upserts the vector
 * into macsync.message_embeddings.
 *
 * Target: p95 < 30s from message insertion to embedding stored.
 *
 * Usage: call startEmbeddingWorker() once at app startup. It runs for the
 * lifetime of the process.
 */

import { Client } from 'pg';
import { logger } from '@/shared/logger';
import { getDb } from '@/shared/db';
import { embedText, EMBEDDING_MODEL_VERSION } from '@/shared/model-boss';
import { upsertEmbedding } from '@/entities/message-embedding';

interface MessageInsertedPayload {
  id: string;
  body: string;
  conversation_id: string;
  device_id: string;
  is_from_me: boolean;
  sent_at: string;
}

function parsePayload(raw: string): MessageInsertedPayload | null {
  try {
    return JSON.parse(raw) as MessageInsertedPayload;
  } catch (err) {
    logger.warn('embedding-worker: failed to parse notification payload', {
      raw: raw.slice(0, 200),
      err: err instanceof Error ? err.message : String(err),
    });
    return null;
  }
}

async function embedMessage(messageId: string, body: string): Promise<void> {
  const trimmed = body.trim();
  if (!trimmed) {
    return;
  }

  const embedding = await embedText(trimmed);
  const pool = getDb();
  await upsertEmbedding(pool, {
    messageId,
    embedding,
    modelVersion: EMBEDDING_MODEL_VERSION,
  });

  logger.debug('embedding-worker: embedded message', { messageId });
}

let listenerClient: Client | null = null;

/**
 * Start the background LISTEN worker.
 *
 * Reconnects automatically on connection failure with exponential backoff.
 * Throws immediately if the db URL is unavailable at boot.
 */
export async function startEmbeddingWorker(): Promise<void> {
  const dbUrl = process.env['QUINN_MACSYNC_DB_URL'];
  if (!dbUrl) {
    throw new Error('QUINN_MACSYNC_DB_URL required for embedding worker');
  }

  logger.info('embedding-worker: starting');
  void connectWithRetry(dbUrl);
}

async function connectWithRetry(dbUrl: string): Promise<void> {
  let backoffMs = 2_000;
  const maxBackoffMs = 60_000;

  while (true) {
    try {
      await runListener(dbUrl);
    } catch (err) {
      logger.error('embedding-worker: listener crashed, reconnecting', {
        err: err instanceof Error ? err.message : String(err),
        backoffMs,
      });
    }

    await new Promise<void>((resolve) => setTimeout(resolve, backoffMs));
    backoffMs = Math.min(backoffMs * 2, maxBackoffMs);
  }
}

async function runListener(dbUrl: string): Promise<void> {
  const client = new Client({ connectionString: dbUrl });
  listenerClient = client;

  await client.connect();
  await client.query('LISTEN message_inserted');

  logger.info('embedding-worker: listening on message_inserted');

  await new Promise<never>((_, reject) => {
    client.on('notification', (msg) => {
      if (msg.channel !== 'message_inserted' || !msg.payload) return;

      const payload = parsePayload(msg.payload);
      if (!payload?.id) return;

      embedMessage(payload.id, payload.body).catch((err) => {
        logger.error('embedding-worker: failed to embed message', {
          messageId: payload.id,
          err: err instanceof Error ? err.message : String(err),
        });
      });
    });

    client.on('error', (err) => {
      reject(err);
    });

    client.on('end', () => {
      reject(new Error('pg client ended unexpectedly'));
    });
  });
}

export function stopEmbeddingWorker(): void {
  if (listenerClient) {
    listenerClient.end().then(
      () => { logger.info('embedding-worker: stopped'); },
      (err: unknown) => { logger.warn('embedding-worker: stop error', { err: String(err) }); },
    );
    listenerClient = null;
  }
}
