/**
 * Centroid-recompute worker.
 *
 * Consumes the 'identity-centroid' queue. For each identity ID:
 *   - Load every face assigned to this identity (match_status='matched').
 *   - Recompute the centroid as the mean of those embeddings.
 *   - Persist centroid_embedding + face_count + centroid_updated_at.
 *
 * Jobs are deduped per identity-id and run after a small delay so a burst of
 * match-worker assignments collapses into a single recompute.
 */

import type { Queue } from 'bullmq';

import { setCentroid } from '@/entities/identity';
import { getDb, pgAll } from '@/shared/db';
import { logger } from '@/shared/logger';
import { createQueue, createWorker } from '@/shared/queues';
import { meanVector } from '@/shared/vector';

const CENTROID_QUEUE_NAME = 'identity-centroid';

export interface CentroidRecomputeJob {
  readonly identityId: string;
}

let centroidQueue: Queue<CentroidRecomputeJob> | null = null;

export function getCentroidQueue(): Queue<CentroidRecomputeJob> {
  if (!centroidQueue) centroidQueue = createQueue<CentroidRecomputeJob>(CENTROID_QUEUE_NAME);
  return centroidQueue;
}

async function processOne(job: CentroidRecomputeJob): Promise<void> {
  try {
    const pool = getDb();
    const rows = await pgAll<{ embedding: readonly number[] }>(
      pool,
      `SELECT embedding FROM macsync.faces WHERE identity_id = $1 AND match_status = 'matched'`,
      [job.identityId],
    );
    if (rows.length === 0) return;

    const centroid = meanVector(rows.map((r) => r.embedding));
    if (centroid.length === 0) return;

    await setCentroid(pool, job.identityId, centroid, rows.length);
    logger.debug('centroid-worker: updated', { identityId: job.identityId, faceCount: rows.length });
  } catch (err) {
    logger.warn('centroid-worker: job error', {
      identityId: job.identityId,
      err: err instanceof Error ? err.message : String(err),
    });
    throw err;
  }
}

let started = false;

export function startCentroidWorker(): void {
  if (started) return;
  started = true;
  getCentroidQueue();
  createWorker<CentroidRecomputeJob>(CENTROID_QUEUE_NAME, async (job) => {
    await processOne(job.data);
  });
}
