/**
 * How a stream learns there is something new: the process tells it.
 *
 * ## The choice, and what was measured before making it
 *
 * The two honest options were polling the table from inside each stream, and an
 * in-process signal. Polling was priced first, against the real database with
 * the whole roster and three months of traffic in it, 9,650 rows: the query a
 * poll would run, `where person_id = $1 and (created_at, id) > (that row's)`
 * off `notifications_person_idx`, executes in 0.042 ms and takes 0.10 ms
 * including the round trip. One tick for every person at once is 13.5 ms.
 *
 * That is not expensive, and it is not the reason. On a five-second interval,
 * 400 connections is 80 queries a second, 6.9 million a day, to deliver the
 * four notifications a day this system sends: the answer is "nothing" 6,999,996
 * times. Each of those is a round trip from Pi 1 to Pi 2 holding one of ten
 * pooled connections, and the loopback figure above is the flattering one. And
 * whatever interval is chosen is also the delay: five seconds of a person
 * staring at a screen that already knows.
 *
 * The signal costs one query, when there is something to say, on the connection
 * that already has something to say, and the latency is a round trip. So: a
 * signal.
 *
 * ## What this deliberately is not
 *
 * It carries no payload. A listener is woken and then reads the table for
 * itself, from the cursor it last sent. That is what makes the live path and
 * the reconnect path the same code: rows are never handed between two places
 * that could disagree about their order, nothing can arrive twice, and nothing
 * can fall into the gap between subscribing and the first query.
 *
 * ## What breaks if a second instance is ever run
 *
 * This reaches the streams held by *this* process and no others. Two Next.js
 * instances behind Caddy would mean a notification written by one is never
 * announced to a person connected to the other: they would see it on their next
 * page load, which is exactly where this system was before SSE, rather than in
 * real time. Nothing is lost and nothing is duplicated, because the row is in
 * the database either way and the stream's cursor is what decides delivery.
 *
 * `architecture.md` §2 specifies one Next.js container on Pi 1, so that is not
 * today's shape. There is one second process already, `scripts/daily.ts`, and
 * it is the same case: what it drains is announced to nobody. It runs at night,
 * when the answer to "who is connected" is nobody, and a stream left open
 * through it hears nothing for at most half an hour: the route closes and
 * remakes every connection on that interval, and the reconnect replays from its
 * cursor. Late, never lost.
 *
 * The seam if a second Next.js instance ever becomes today's shape is this file
 * alone: `publish` becomes `pg_notify` and `subscribe` becomes a `LISTEN` on
 * the connection `postgres.js` already offers, and no caller changes.
 */

/** Woken, not told. The listener reads the database for itself. */
type Wake = () => void

/**
 * Pinned to `globalThis` for the reason lib/db/client.ts pins the pool: the
 * development server re-evaluates a module on every edit, and a second copy of
 * this map is a set of live streams nothing can reach any more. It also settles
 * a subtler question in production, where a Server Action and a Route Handler
 * are compiled into different bundles and a module-level constant is not
 * necessarily one object. Here it always is.
 */
declare global {
  var __advmfrNotifyHub: Map<string, Set<Wake>> | undefined
}

const listeners: Map<string, Set<Wake>> = (globalThis.__advmfrNotifyHub ??=
  new Map())

/**
 * Listen for this person, and be handed the way to stop.
 *
 * The returned function is the only way out, and calling it twice is safe: a
 * stream is torn down from an abort, from a cancel and from a timer, and all
 * three run.
 */
export function subscribe(personId: string, wake: Wake): () => void {
  let set = listeners.get(personId)
  if (!set) {
    set = new Set()
    listeners.set(personId, set)
  }
  set.add(wake)

  return () => {
    const current = listeners.get(personId)
    if (!current) return
    current.delete(wake)
    // The map holds one entry per person with a tab open, not one per person in
    // the roster. An empty set left behind is a small leak that never ends.
    if (current.size === 0) listeners.delete(personId)
  }
}

/**
 * Something was written for this person.
 *
 * Over a copy of the set, because a listener that fails and tears itself down
 * would otherwise be removed from the set being iterated. Failures are caught
 * per listener: one dead stream must not stop the person's other three tabs
 * from hearing about the same row.
 */
export function publish(personId: string): void {
  const set = listeners.get(personId)
  if (!set) return
  for (const wake of [...set]) {
    try {
      wake()
    } catch (error) {
      console.error(
        `A notification stream for ${personId} threw when woken. It stays subscribed until its own cleanup runs:`,
        error,
      )
    }
  }
}

/**
 * How many streams are listening, for one person or for everybody.
 *
 * Exported because "the stream stopped" is not something a test can see from
 * the outside: a closed socket says the client went away, not that the server
 * let go of it. This is the number that has to come back to zero.
 */
export function listenerCount(personId?: string): number {
  if (personId !== undefined) return listeners.get(personId)?.size ?? 0
  let total = 0
  for (const set of listeners.values()) total += set.size
  return total
}