import { afterEach, beforeEach, describe, expect, test } from 'bun:test';
import { mkdtempSync, rmSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import type { InboundMessage } from '@celilo/capabilities';
import { eq } from 'drizzle-orm';
import type { DbClient } from '../../db/client';
import { type Route, alerts, modules, monitors, notificationDeliveries } from '../../db/schema';
import { setupTestDatabaseAt } from '../../test-utils/database';
import { resetTestDbPath } from '../../test-utils/db-path';
import {
  type InboundPollDeps,
  type TransportReadRecord,
  pollInbound,
  transportsWithRoutes,
} from './inbound-poller';
import { moduleCheckAlertKey } from './keys';
import { createPerson, createRoute } from './people';
import { mintDelivery } from './tokens';

const NOW = new Date('2026-07-28T03:00:00Z');
const KEY = moduleCheckAlertKey('caddy', 'disk-space');
const PETER = '+15550001';
const WIFE = '+15550002';
const STRANGER = '+15559999';

describe('pollInbound', () => {
  let dir: string;
  let db: DbClient;
  let peterRoute: Route;
  let wifeRoute: Route;
  const cursors = new Map<string, string | null>();

  function deps(messages: InboundMessage[], over: Partial<InboundPollDeps> = {}): InboundPollDeps {
    return {
      receiveFrom: async () => ({ status: 'received', messages, cursor: 'cursor-1' }),
      readCursor: (t) => cursors.get(t) ?? null,
      writeCursor: (t, c) => {
        cursors.set(t, c);
      },
      now: () => NOW,
      ...over,
    };
  }

  const inbound = (from: string, body: string): InboundMessage => ({
    senderAddress: from,
    body,
    receivedAt: NOW.getTime(),
    messageId: `${from}:${NOW.getTime()}`,
  });

  beforeEach(async () => {
    dir = mkdtempSync(join(tmpdir(), 'inbound-poll-'));
    const dbPath = join(dir, 'celilo.db');
    process.env.CELILO_DB_PATH = dbPath;
    db = await setupTestDatabaseAt(dbPath);
    cursors.clear();

    db.insert(modules)
      .values({
        id: 'signal',
        name: 'Signal',
        version: '0.1.0',
        manifestData: {},
        sourcePath: '/tmp/signal',
      })
      .run();
    db.insert(monitors)
      .values({
        id: 'mon-1',
        kind: 'module_hook',
        target: 'caddy',
        intervalMinutes: 15,
        severity: 'critical',
      })
      .run();
    db.insert(alerts)
      .values({
        id: 'alert-1',
        key: KEY,
        activeKey: KEY,
        monitorId: 'mon-1',
        state: 'firing',
        severity: 'critical',
        graceUntil: NOW,
        message: '/var 94% used',
      })
      .run();

    const peter = createPerson(db, { name: 'peter', timezone: 'UTC' });
    const wife = createPerson(db, { name: 'wife', timezone: 'UTC' });
    peterRoute = createRoute(db, {
      personId: peter.id,
      transportModuleId: 'signal',
      address: PETER,
      canAck: true,
    });
    wifeRoute = createRoute(db, {
      personId: wife.id,
      transportModuleId: 'signal',
      address: WIFE,
      canAck: true,
    });
  });

  afterEach(() => {
    db.$client.close();
    resetTestDbPath();
    try {
      rmSync(dir, { recursive: true, force: true });
    } catch {
      /* ignore */
    }
  });

  const page = (routeId: string) =>
    mintDelivery(db, {
      kind: 'alert',
      targetId: 'alert-1',
      routeId,
      now: NOW,
      ttlMs: 86_400_000,
    });

  const alertState = () => db.select().from(alerts).where(eq(alerts.id, 'alert-1')).get()?.state;
  const ackedBy = () => db.select().from(alerts).where(eq(alerts.id, 'alert-1')).get()?.ackedBy;

  test('only transports with routes are polled', () => {
    expect(transportsWithRoutes(db)).toEqual(['signal']);
  });

  test('a valid token from the right sender acknowledges the alert', async () => {
    const delivery = page(peterRoute.id);
    const report = await pollInbound(db, deps([inbound(PETER, delivery.token)]));

    expect(report.acked).toBe(1);
    expect(alertState()).toBe('acked');
  });

  // 7.5 — the secondary acks, and the primary is told so they can stand down.
  test('an ack is broadcast to everyone else who was paged', async () => {
    const sent: { address: string; body: string }[] = [];
    page(peterRoute.id);
    const wifeDelivery = page(wifeRoute.id);

    const report = await pollInbound(
      db,
      deps([inbound(WIFE, wifeDelivery.token)], {
        transportFor: () => ({
          send: async (request: { address: string; body: string }) => {
            sent.push(request);
            return { messageId: 'm-1' };
          },
        }),
      }),
    );

    expect(report.acked).toBe(1);
    expect(report.broadcast).toBe(1);
    expect(sent).toHaveLength(1);
    expect(sent[0].address).toBe(PETER);
    expect(sent[0].body).toContain('wife');
  });

  test('the acknowledger is not told about their own ack', async () => {
    const sent: string[] = [];
    const delivery = page(peterRoute.id);

    await pollInbound(
      db,
      deps([inbound(PETER, delivery.token)], {
        transportFor: () => ({
          send: async (request: { address: string; body: string }) => {
            sent.push(request.address);
            return { messageId: 'm-1' };
          },
        }),
      }),
    );

    expect(sent).toEqual([]);
  });

  // An installation with no send path still acks — it just cannot tell anyone.
  test('a missing transport does not stop the ack', async () => {
    page(peterRoute.id);
    const wifeDelivery = page(wifeRoute.id);
    const report = await pollInbound(db, deps([inbound(WIFE, wifeDelivery.token)]));

    expect(report.acked).toBe(1);
    expect(report.broadcast).toBe(0);
    expect(alertState()).toBe('acked');
  });

  test('a token from the wrong sender is rejected and changes nothing', async () => {
    const delivery = page(peterRoute.id);
    const report = await pollInbound(db, deps([inbound(WIFE, delivery.token)]));

    expect(report.unheard).toEqual([{ senderAddress: WIFE, reason: 'wrong_sender' }]);
    expect(report.acked).toBe(0);
    expect(alertState()).toBe('firing');
  });

  // Silence, not a reply — answering confirms the number is live.
  test('an unknown sender is ignored', async () => {
    const delivery = page(peterRoute.id);
    const report = await pollInbound(db, deps([inbound(STRANGER, delivery.token)]));

    expect(report.unheard).toEqual([{ senderAddress: STRANGER, reason: 'unknown_sender' }]);
    expect(alertState()).toBe('firing');
  });

  test('the token is consumed so it cannot be replayed', async () => {
    const delivery = page(peterRoute.id);
    await pollInbound(db, deps([inbound(PETER, delivery.token)]));

    const row = db
      .select()
      .from(notificationDeliveries)
      .where(eq(notificationDeliveries.id, delivery.id))
      .get();
    expect(row?.consumedAt).not.toBeNull();
  });

  // At-least-once is the contract: the same batch replayed after a crash must
  // not do anything twice that matters.
  test('replaying the same message is harmless', async () => {
    const delivery = page(peterRoute.id);
    const message = inbound(PETER, delivery.token);

    const first = await pollInbound(db, deps([message]));
    const second = await pollInbound(db, deps([message]));

    expect(first.acked).toBe(1);
    // The token is spent, so the replay is refused rather than acking again.
    expect(second.acked).toBe(0);
    expect(second.unheard).toEqual([{ senderAddress: PETER, reason: 'unknown_token' }]);
    expect(alertState()).toBe('acked');
  });

  test('a bare ack works when exactly one delivery is outstanding', async () => {
    page(peterRoute.id);
    const report = await pollInbound(db, deps([inbound(PETER, 'ack')]));
    expect(report.acked).toBe(1);
    expect(alertState()).toBe('acked');
  });

  test('a bare ack is refused when several are outstanding', async () => {
    page(peterRoute.id);
    db.insert(alerts)
      .values({
        id: 'alert-2',
        key: 'module:caddy/check:cert',
        activeKey: 'module:caddy/check:cert',
        monitorId: 'mon-1',
        state: 'firing',
        severity: 'critical',
        graceUntil: NOW,
        message: 'expiring',
      })
      .run();
    mintDelivery(db, {
      kind: 'alert',
      targetId: 'alert-2',
      routeId: peterRoute.id,
      now: NOW,
      ttlMs: 86_400_000,
    });

    const report = await pollInbound(db, deps([inbound(PETER, 'ack')]));
    expect(report.unheard).toEqual([{ senderAddress: PETER, reason: 'ambiguous' }]);
    expect(alertState()).toBe('firing');
  });

  // One dead transport must not stop the others being read.
  test('an unreachable transport is skipped without throwing', async () => {
    const report = await pollInbound(
      db,
      deps([], { receiveFrom: async () => ({ status: 'failed', error: 'connection refused' }) }),
    );
    expect(report.transportsPolled).toBe(1);
    expect(report.messagesRead).toBe(0);
  });

  /**
   * The regression that cost a week. signal-cli's daemon refused celilo's read
   * call outright ("Receive command cannot be used if messages are already
   * being received") and the poller reported the same "0 message(s)" it
   * reports when nobody has replied. The reason must reach the report, or the
   * two are indistinguishable from outside.
   */
  test('a transport that REFUSES the read says so instead of looking quiet', async () => {
    const refusal = 'Receive command cannot be used if messages are already being received.';
    const report = await pollInbound(
      db,
      deps([], { receiveFrom: async () => ({ status: 'failed', error: refusal }) }),
    );

    expect(report.failures).toEqual([{ transportModuleId: 'signal', error: refusal }]);
    // ...and a genuinely quiet transport must NOT look like a failure.
    const quiet = await pollInbound(db, deps([]));
    expect(quiet.failures).toEqual([]);
    expect(quiet.messagesRead).toBe(0);
  });

  test('a unidirectional transport is not reported as a failure', async () => {
    const report = await pollInbound(
      db,
      deps([], { receiveFrom: async () => ({ status: 'unidirectional' }) }),
    );
    expect(report.failures).toEqual([]);
    expect(report.transportsPolled).toBe(1);
  });

  test('the cursor is persisted after a successful poll', async () => {
    await pollInbound(db, deps([]));
    expect(cursors.get('signal')).toBe('cursor-1');
  });

  test('the cursor is NOT advanced when the transport could not be reached', async () => {
    await pollInbound(
      db,
      deps([], { receiveFrom: async () => ({ status: 'failed', error: 'connection refused' }) }),
    );
    expect(cursors.get('signal')).toBeUndefined();
  });

  test('gibberish is rejected without touching the alert', async () => {
    page(peterRoute.id);
    const report = await pollInbound(db, deps([inbound(PETER, 'what is going on')]));
    expect(report.unheard).toEqual([{ senderAddress: PETER, reason: 'unrecognised' }]);
    expect(alertState()).toBe('firing');
  });

  // The seam that was untested. `acknowledgeAlert` returning null was covered
  // (ack.test.ts), and pollInbound was covered — but not what the REPORT says
  // when the two meet. Observed live: a reply produced `1 acked` while every
  // alert still read `ackedBy: null`, because the counter incremented next to
  // the call instead of observing its result.
  describe('a token whose alert no longer exists', () => {
    test('is NOT counted as acked', async () => {
      const delivery = page(peterRoute.id);
      db.delete(alerts).run(); // the alert this token names is gone

      const report = await pollInbound(db, deps([inbound(PETER, delivery.token)]));

      expect(report.acked).toBe(0);
    });

    test('is reported as stale_target rather than silently', async () => {
      const delivery = page(peterRoute.id);
      db.delete(alerts).run();

      const report = await pollInbound(db, deps([inbound(PETER, delivery.token)]));

      expect(report.unheard).toEqual([{ senderAddress: PETER, reason: 'stale_target' }]);
    });

    // The token is still single-use: a dead target must not leave it replayable.
    test('still consumes the token', async () => {
      const delivery = page(peterRoute.id);
      db.delete(alerts).run();

      await pollInbound(db, deps([inbound(PETER, delivery.token)]));
      const second = await pollInbound(db, deps([inbound(PETER, delivery.token)]));

      expect(second.acked).toBe(0);
      expect(second.unheard).toEqual([{ senderAddress: PETER, reason: 'unknown_token' }]);
    });

    // Nobody should be told "someone took it" when nobody did.
    test('broadcasts nothing', async () => {
      const delivery = page(peterRoute.id);
      db.delete(alerts).run();

      const report = await pollInbound(db, deps([inbound(PETER, delivery.token)]));

      expect(report.broadcast).toBe(0);
    });
  });

  // The counter must track the database, not the attempt. If these two ever
  // disagree again, the operator sees a success that did not happen.
  test('report.acked agrees with what the alert row actually says', async () => {
    const delivery = page(peterRoute.id);

    const report = await pollInbound(db, deps([inbound(PETER, delivery.token)]));

    expect(report.acked).toBe(1);
    expect(alertState()).toBe('acked');
    expect(ackedBy()).toBe(peterRoute.personId);
  });

  // The gap this closes. The ONLY per-transport state celilo persisted was the
  // cursor, and writeCursor early-returns when there is no cursor — which a
  // failed read never produces. So the store could not REPRESENT a failure, and
  // "unreadable for a week" and "nobody replied for a week" left identical
  // traces (#501).
  describe('recording what each read attempt produced', () => {
    const records: Array<[string, TransportReadRecord]> = [];
    const recording = (over: Partial<InboundPollDeps> = {}) =>
      deps([], { recordRead: (t, r) => records.push([t, r]), ...over });

    beforeEach(() => {
      records.length = 0;
    });

    test('a FAILED read is recorded — the case the cursor could never express', async () => {
      await pollInbound(
        db,
        recording({ receiveFrom: async () => ({ status: 'failed', error: 'connection refused' }) }),
      );
      expect(records).toHaveLength(1);
      expect(records[0][0]).toBe('signal');
      expect(records[0][1]).toMatchObject({ outcome: 'failed', error: 'connection refused' });
    });

    test('a successful read is recorded with how many messages it returned', async () => {
      await pollInbound(
        db,
        deps([inbound(PETER, 'hello')], { recordRead: (t, r) => records.push([t, r]) }),
      );
      expect(records[0][1]).toMatchObject({ outcome: 'received', messages: 1 });
    });

    // Zero messages is not failure. Conflating them is the original bug.
    test('an empty but successful read records received, not failed', async () => {
      await pollInbound(db, recording());
      expect(records[0][1]).toMatchObject({ outcome: 'received', messages: 0 });
    });

    test('a unidirectional transport is recorded as such', async () => {
      await pollInbound(db, recording({ receiveFrom: async () => ({ status: 'unidirectional' }) }));
      expect(records[0][1].outcome).toBe('unidirectional');
    });

    // Every attempt, not just interesting ones — a gap in the record would be
    // read as "nothing happened".
    test('every attempt is recorded, and carries when it happened', async () => {
      await pollInbound(db, recording());
      expect(records[0][1].at).toBe(NOW.toISOString());
    });

    // Optional dep: a caller that does not persist must still poll.
    test('polling works with no recorder attached', async () => {
      const report = await pollInbound(db, deps([]));
      expect(report.transportsPolled).toBe(1);
    });
  });

  test('a route pointing at a transport nobody uses is not polled', () => {
    db.delete(alerts).run();
    expect(transportsWithRoutes(db)).toEqual(['signal']);
    expect(wifeRoute.transportModuleId).toBe('signal');
  });

  /**
   * Answering a deploy question, at the SEAM.
   *
   * This is the coverage #533 was missing. `parseInterviewAnswer` was unit
   * tested and correct; `pollInbound` was unit tested and correct; and an
   * answer sent in the shape the page asks for was rejected as `unrecognised`
   * for months, because nothing exercised an interview delivery THROUGH the
   * poller. Two green units, wrong wiring — the same shape as the ack counter
   * above.
   */
  describe('an answer to a deploy question', () => {
    const ask = (routeId: string, eventId = '42') =>
      mintDelivery(db, {
        kind: 'interview',
        targetId: eventId,
        routeId,
        now: NOW,
        ttlMs: 60_000,
      });

    test('is published against the waiting question and counted as answered', async () => {
      const question = ask(peterRoute.id, '42');
      const answered: { eventId: string; value: string }[] = [];

      const report = await pollInbound(
        db,
        deps([inbound(PETER, `${question.token} admin@example.org`)], {
          answerInterview: (eventId, value) => answered.push({ eventId, value }),
        }),
      );

      expect(answered).toEqual([{ eventId: '42', value: 'admin@example.org' }]);
      expect(report.answered).toBe(1);
      // An answer is not an acknowledgement; nothing about an alert moved.
      expect(report.acked).toBe(0);
      expect(report.unheard).toEqual([]);
    });

    test('a value containing spaces survives verbatim', async () => {
      const question = ask(peterRoute.id);
      const answered: string[] = [];

      await pollInbound(
        db,
        deps([inbound(PETER, `${question.token} my value: with, punctuation`)], {
          answerInterview: (_id, value) => answered.push(value),
        }),
      );

      expect(answered).toEqual(['my value: with, punctuation']);
    });

    test('the token is consumed, so an answer cannot be replayed', async () => {
      const question = ask(peterRoute.id);
      const answered: string[] = [];
      const withResponder = (body: string) =>
        pollInbound(
          db,
          deps([inbound(PETER, body)], { answerInterview: (_id, v) => answered.push(v) }),
        );

      await withResponder(`${question.token} first`);
      const second = await withResponder(`${question.token} second`);

      expect(answered).toEqual(['first']);
      expect(second.answered).toBe(0);
      // `unrecognised`, not `unknown_token`: once the token no longer resolves,
      // `<word> <text>` is indistinguishable from an ordinary sentence, so
      // celilo declines to assert the operator mistyped a token.
      expect(second.unheard).toEqual([{ senderAddress: PETER, reason: 'unrecognised' }]);
    });

    // Naming the question but supplying nothing is its own answer, and the
    // token must survive so the operator's next attempt can work.
    test('a token with no value reports needs_value and leaves the token live', async () => {
      const question = ask(peterRoute.id);

      const first = await pollInbound(
        db,
        deps([inbound(PETER, question.token)], { answerInterview: () => {} }),
      );
      expect(first.answered).toBe(0);
      expect(first.unheard).toEqual([{ senderAddress: PETER, reason: 'needs_value' }]);

      const answered: string[] = [];
      const second = await pollInbound(
        db,
        deps([inbound(PETER, `${question.token} admin@example.org`)], {
          answerInterview: (_id, v) => answered.push(v),
        }),
      );
      expect(second.answered).toBe(1);
      expect(answered).toEqual(['admin@example.org']);
    });

    // With no responder attached there is nothing to publish against, so the
    // question stays unanswered AND the token stays usable.
    test('with no responder attached the token is not burned', async () => {
      const question = ask(peterRoute.id);

      const report = await pollInbound(db, deps([inbound(PETER, `${question.token} value`)]));
      expect(report.answered).toBe(0);
      expect(report.unheard).toEqual([{ senderAddress: PETER, reason: 'unrecognised' }]);

      const answered: string[] = [];
      const retry = await pollInbound(
        db,
        deps([inbound(PETER, `${question.token} value`)], {
          answerInterview: (_id, v) => answered.push(v),
        }),
      );
      expect(retry.answered).toBe(1);
      expect(answered).toEqual(['value']);
    });

    test('an answer from a number the question was not sent to is refused', async () => {
      const question = ask(peterRoute.id);
      const answered: string[] = [];

      const report = await pollInbound(
        db,
        deps([inbound(WIFE, `${question.token} admin@example.org`)], {
          answerInterview: (_id, v) => answered.push(v),
        }),
      );

      expect(answered).toEqual([]);
      expect(report.unheard).toEqual([{ senderAddress: WIFE, reason: 'wrong_sender' }]);
    });
  });
});