import { and, count, desc, eq, inArray, isNull, sql } from "drizzle-orm"
import { alias } from "drizzle-orm/pg-core"

import { db } from "@/lib/db/client"
import {
  domainEvents,
  notifications,
  people,
  recurringTemplates,
  taskInstances,
  type DomainEventType,
} from "@/lib/db/schema"
import type { NoticeKey } from "@/lib/notice"
import { publish } from "@/lib/services/notification-hub"
import type { SignedIn } from "@/lib/services/session"

/**
 * Telling somebody, which is the only half of the outbox that has a reader
 * today.
 *
 * ## The seam, and where this file sits on it
 *
 * `architecture.md` §4: application code never talks to a channel. It writes an
 * event (lib/services/events.ts) and stops. Everything below pulls from that
 * table and decides who hears about it. Replacing the puller with a pg-boss
 * worker changes `dispatchPending` and its two call sites and nothing else: not
 * a service that changes a task, not the fan-out rules, not the drivers.
 *
 * ## One deviation from the interface as the specification writes it
 *
 * §4 gives the driver as `send(userId, { title, body, taskId? })`, with `title`
 * and `body` as strings. `title` here is a dictionary key rather than a
 * sentence, and that is deliberate: 183 of the 193 people read Thai, and a
 * title rendered by the process that sent it is a title that can only ever be
 * read in the language that process happened to be thinking in. A driver that
 * needs words has the reader in hand and can ask the dictionary for them; a
 * driver handed words has lost the ability to choose them. `body` stays a
 * string, because it is what somebody typed: data, not interface, and never
 * translated.
 */

/* -------------------------------------------------------------------------- */
/* The driver                                                                 */
/* -------------------------------------------------------------------------- */

/** One thing to tell one person. */
export type Notice = {
  /** Which event this fans out from, and the half of the key that makes a repeat harmless. */
  eventId: string
  /** The sentence, named. See lib/notice.ts. */
  title: NoticeKey
  /** The words the actor wrote, when the act carried any. */
  body?: string | null
  /** What it is about, and what a row links to. */
  taskId?: string | null
}

/**
 * The seam `architecture.md` §4 specifies, with the one adjustment above.
 *
 * `userId` there is `personId` here: this system's notifications are addressed
 * to people in the roster, and 190 of the 193 have no login at all. A driver
 * that needs an account looks it up.
 */
export interface NotificationDriver {
  send(personId: string, n: Notice): Promise<void>
}

/**
 * The only implementation there is: a row somebody reads in the app, and a word
 * to any stream they have open.
 *
 * Idempotent by the unique index on (event_id, person_id) rather than by
 * checking first, because checking first is a race and the index is not.
 *
 * `architecture.md` §4 writes this driver as "DB row + SSE", and the two halves
 * are here in that order, joined by `returning`: the announcement is made only
 * when a row was actually written. A repeat send, the drain re-finding an
 * event a crash left owed, conflicts and returns nothing and says nothing, so
 * the unique index that makes delivery safe to retry makes the stream safe to
 * retry too. It is a wake-up and not the row itself, and
 * lib/services/notification-hub.ts says why.
 */
export const inAppDriver: NotificationDriver = {
  async send(personId, notice) {
    const written = await db
      .insert(notifications)
      .values({
        personId,
        eventId: notice.eventId,
        messageKey: notice.title,
        taskInstanceId: notice.taskId ?? null,
      })
      .onConflictDoNothing({
        target: [notifications.eventId, notifications.personId],
      })
      .returning({ id: notifications.id })

    if (written.length > 0) publish(personId)
  },
}

/**
 * Every channel that is on.
 *
 * `NtfyDriver` is not written and there is no ntfy server on the LAN to write
 * it against, so this list has one member. It is a list rather than a constant
 * so that adding one is adding one line here: the loop below already sends to
 * all of them, and nothing above it knows how many there are.
 *
 * A second driver will want two things this half does not have: preferences,
 * so a person can turn a channel off, and per-channel delivery state, because
 * an HTTP POST cannot be made idempotent by a unique index the way an insert
 * can. Both are the next piece of work rather than a gap here.
 */
const DRIVERS: NotificationDriver[] = [inAppDriver]

/* -------------------------------------------------------------------------- */
/* Who hears about what                                                       */
/* -------------------------------------------------------------------------- */

/**
 * The four notifications v1 sends, all person to person, all low volume, all
 * wanted by somebody who is waiting on the other person.
 *
 * `task.missed` is the one event that is recorded and told to nobody. There are
 * 267 missed instances in the database today and the nightly sweep marks more
 * every night; fanning them out would put a hundred rows a week in front of
 * people whose four real notifications would be underneath them by the first
 * morning. The event is still written, because `architecture.md` §5 names it
 * and a webhook or a KPI consumer may well want it, and the null below is the
 * whole of "notify nobody".
 */
const FAN_OUT: Record<
  DomainEventType,
  { to: "assignee" | "checker"; title: NoticeKey } | null
> = {
  // Somebody is now waiting on the checker, and nothing else tells them.
  "task.finished": { to: "checker", title: "notify.finished" },
  // It carries a note explaining what has to be corrected.
  "task.sent_back": { to: "assignee", title: "notify.sentBack" },
  // It carries a note, and it is the whole point of the button.
  "task.reminded": { to: "assignee", title: "notify.reminded" },
  // The loop closes, and the assignee is the one who was waiting for it.
  "task.checked": { to: "assignee", title: "notify.approved" },
  "task.missed": null,
}

/**
 * An event with the two people its task names, which is everything the fan-out
 * rule needs. Exported with `recipientsOf` below: the rule about who hears what
 * is the one piece of this file with no database in it, and it is worth being
 * able to state the whole matrix without one, exactly as `permits` is in
 * lib/services/task-commands.ts.
 */
export type Pending = {
  id: string
  type: DomainEventType
  taskInstanceId: string | null
  actorPersonId: string | null
  payload: { note?: string } | null
  assigneePersonId: string | null
  checkerPersonId: string | null
}

/**
 * The event, with the two people a task names on it.
 *
 * Read here and not stored on the event, because who checks a duty is a fact
 * about the position and can change between the act and the delivery. The
 * recipient is worked out at the moment somebody is told, which is the moment
 * it has to be right.
 */
function pendingSelect() {
  return db
    .select({
      id: domainEvents.id,
      type: domainEvents.type,
      taskInstanceId: domainEvents.taskInstanceId,
      actorPersonId: domainEvents.actorPersonId,
      payload: domainEvents.payload,
      assigneePersonId: taskInstances.assigneePersonId,
      checkerPersonId: recurringTemplates.checkerPersonId,
    })
    .from(domainEvents)
    .leftJoin(taskInstances, eq(taskInstances.id, domainEvents.taskInstanceId))
    .leftJoin(
      recurringTemplates,
      eq(recurringTemplates.id, taskInstances.recurringTemplateId),
    )
}

/**
 * Who is told about this event, and what they are told.
 *
 * Two rules, and they are both refusals. A recipient nobody holds (a vacant
 * position has no assignee, a general duty has no checker) is not a recipient.
 * And **the actor is never told about their own act**: the person who pressed
 * the button knows. That second one is not hypothetical here, it is the normal
 * case in the small departments, where the manager who approves the work is
 * often the person who did it.
 */
export function recipientsOf(
  event: Pending,
): { personId: string; notice: Notice }[] {
  const rule = FAN_OUT[event.type]
  if (!rule) return []

  const personId =
    rule.to === "checker" ? event.checkerPersonId : event.assigneePersonId
  if (!personId) return []
  if (personId === event.actorPersonId) return []

  return [
    {
      personId,
      notice: {
        eventId: event.id,
        title: rule.title,
        body: event.payload?.note ?? null,
        taskId: event.taskInstanceId,
      },
    },
  ]
}

/** Every driver, for every recipient. Returns how many sends were attempted. */
async function deliver(event: Pending): Promise<number> {
  const recipients = recipientsOf(event)
  for (const { personId, notice } of recipients) {
    for (const driver of DRIVERS) {
      await driver.send(personId, notice)
    }
  }
  return recipients.length
}

/**
 * Mark the event fanned out.
 *
 * `IS NULL` in the WHERE so two drains cannot both claim it and report it. The
 * stamp is bookkeeping and not the guarantee: a process that dies between
 * `deliver` and here leaves the event owed, the next drain sends it again, and
 * the unique index on the notification is what makes that second send change
 * nothing.
 */
async function markDispatched(eventId: string): Promise<void> {
  await db
    .update(domainEvents)
    .set({ dispatchedAt: new Date() })
    .where(and(eq(domainEvents.id, eventId), isNull(domainEvents.dispatchedAt)))
}

/**
 * One event, now, straight after the transaction that wrote it committed.
 *
 * This is what makes the app feel immediate without a worker: the screen that
 * comes back from an approval already has the notification behind it. It is
 * best effort and nothing depends on it — if it throws, the event is still in
 * the outbox with `dispatched_at` null and the nightly drain finds it.
 */
export async function dispatchEvent(eventId: string): Promise<number> {
  const [event] = await pendingSelect().where(
    and(eq(domainEvents.id, eventId), isNull(domainEvents.dispatchedAt)),
  )
  if (!event) return 0
  const sent = await deliver(event)
  await markDispatched(eventId)
  return sent
}

/**
 * The same, for a caller that must not fail because a channel did.
 *
 * Every command calls this and none of them await an outcome they could act on:
 * a manager whose approval went through is not shown an error because a
 * notification did not. The failure is written where somebody operating the
 * system can find it, and the event stays owed.
 */
export async function dispatchAfterCommit(eventId: string): Promise<void> {
  try {
    await dispatchEvent(eventId)
  } catch (error) {
    console.error(
      `The event ${eventId} was recorded and not dispatched. It stays in the outbox:`,
      error,
    )
  }
}

/**
 * Everything still owed, oldest first. What a worker would do, done by whatever
 * calls it: the nightly job today.
 *
 * **Safe to run twice, and that is measured rather than asserted.** A second
 * run finds nothing pending because the first stamped it, and a run that
 * overlaps one that crashed mid-way re-sends what it re-finds into the unique
 * index, which absorbs it.
 *
 * The limit exists so that a drain after a long outage is a series of bounded
 * pieces of work rather than one that holds a connection for a minute. The
 * caller runs it again if it came back full.
 */
export async function dispatchPending(
  limit = 500,
): Promise<{ events: number; sent: number }> {
  const pending = await pendingSelect()
    .where(isNull(domainEvents.dispatchedAt))
    .orderBy(domainEvents.occurredAt)
    .limit(limit)

  let sent = 0
  for (const event of pending) {
    sent += await deliver(event)
    await markDispatched(event.id)
  }
  return { events: pending.length, sent }
}

/* -------------------------------------------------------------------------- */
/* What the person reads                                                      */
/* -------------------------------------------------------------------------- */

export type NotificationRow = {
  id: string
  /** The sentence, named. The screen asks the dictionary in the reader's language. */
  key: NoticeKey
  /**
   * The one value that sentence takes: who did it, spelled as the roster spells
   * them. A name is data and is never translated.
   */
  name: string
  /** What they wrote, when the act carried words. Shown as theirs, not as ours. */
  note: string | null
  /** The task to open. Null only if a future event type carries none. */
  taskId: string | null
  /** What that task is, so the row says which work it is about. */
  task: string | null
  at: Date
  read: boolean
}

const actor = alias(people, "actor")

/**
 * How far back the list goes.
 *
 * Enough that nothing a person could reasonably still act on falls off, and
 * bounded because this is one query on a shared terminal. There is no paging:
 * a notification older than the last hundred has been overtaken by the task
 * itself, which is where the work actually lives.
 */
const FEED_LIMIT = 100

export async function notificationFeed(
  user: SignedIn,
): Promise<NotificationRow[]> {
  const rows = await db
    .select({
      id: notifications.id,
      key: notifications.messageKey,
      createdAt: notifications.createdAt,
      readAt: notifications.readAt,
      taskId: notifications.taskInstanceId,
      note: domainEvents.payload,
      firstName: actor.firstName,
      lastName: actor.lastName,
      displayName: actor.displayName,
      task: recurringTemplates.description,
    })
    .from(notifications)
    .innerJoin(domainEvents, eq(domainEvents.id, notifications.eventId))
    // The actor is a left join although every notice v1 sends has one: an event
    // with no person behind it is a real state (the nightly sweep), and an
    // inner join would silently drop such a row rather than show it.
    .leftJoin(actor, eq(actor.id, domainEvents.actorPersonId))
    .leftJoin(taskInstances, eq(taskInstances.id, notifications.taskInstanceId))
    .leftJoin(
      recurringTemplates,
      eq(recurringTemplates.id, taskInstances.recurringTemplateId),
    )
    .where(eq(notifications.personId, user.personId))
    // Unread first, then newest. Both, and in that order, because the list is
    // opened to answer "what is new" and read again to answer "what was it".
    .orderBy(
      sql`${notifications.readAt} IS NULL DESC`,
      desc(notifications.createdAt),
    )
    .limit(FEED_LIMIT)

  return rows.map((r) => ({
    id: r.id,
    key: r.key,
    name: r.displayName ?? `${r.firstName ?? ""} ${r.lastName ?? ""}`.trim(),
    note: r.note?.note ?? null,
    taskId: r.taskId,
    task: r.task,
    at: r.createdAt,
    read: r.readAt !== null,
  }))
}

/** What the sidebar puts on the entry. One indexed count. */
export async function unreadCount(personId: string): Promise<number> {
  const [row] = await db
    .select({ n: count() })
    .from(notifications)
    .where(
      and(eq(notifications.personId, personId), isNull(notifications.readAt)),
    )
  return row?.n ?? 0
}

/* -------------------------------------------------------------------------- */
/* What a stream sends                                                        */
/* -------------------------------------------------------------------------- */

/**
 * How many rows one call may answer with.
 *
 * A live stream sends one row per wake-up and never comes near this. It exists
 * for the reconnect after a long absence, which is the case that could
 * otherwise hand a browser a morning's backlog in one burst; the caller asks
 * again if it came back full, exactly as the drain does. Exported because that
 * is the contract: a full batch means there is more.
 */
export const STREAM_BATCH = 50

/**
 * The person's newest notification, which is where a fresh stream starts.
 *
 * Null when they have none, and that null is the reason there is no synthetic
 * "beginning" cursor. A stream with nothing behind it sends no `id:` at all, so
 * the browser reconnects with no `Last-Event-ID`, so it lands back here and
 * asks the same question. What arrived in between is everything, and everything
 * is what it gets.
 */
export async function newestNotificationId(
  personId: string,
): Promise<string | null> {
  const [row] = await db
    .select({ id: notifications.id })
    .from(notifications)
    .where(eq(notifications.personId, personId))
    .orderBy(desc(notifications.createdAt), desc(notifications.id))
    .limit(1)
  return row?.id ?? null
}

/**
 * Everything this person was sent after that one, oldest first.
 *
 * ## Why the comparison is a tuple, and why it is done in SQL
 *
 * `created_at` alone is not an order. Two events dispatched in the same
 * millisecond by the same drain loop are two rows this would have to choose
 * between, and choosing wrongly means one of them is skipped for good rather
 * than delivered late. `(created_at, id)` is a total order, and the id is
 * already unique.
 *
 * The anchor is read inside the statement rather than passed in as a time. A
 * `timestamptz` holds microseconds and a JavaScript `Date` holds milliseconds,
 * so a cursor that made the trip through one would name an instant a fraction
 * before the row it came from, and that row would be re-sent on every single
 * reconnect. See lib/notification-cursor.ts.
 *
 * An `after` that names no row makes the comparison null and the answer empty,
 * which is the right reading of a cursor from a database that has since been
 * reseeded: start from here, replay nothing. A null `after` is the other case
 * entirely and means everything, capped.
 */
export async function notificationsSince(
  personId: string,
  after: string | null,
  limit = STREAM_BATCH,
): Promise<string[]> {
  const rows = await db
    .select({ id: notifications.id })
    .from(notifications)
    .where(
      and(
        eq(notifications.personId, personId),
        after
          ? sql`(${notifications.createdAt}, ${notifications.id}) > (select created_at, id from ${notifications} where id = ${after})`
          : undefined,
      ),
    )
    .orderBy(notifications.createdAt, notifications.id)
    .limit(limit)

  return rows.map((r) => r.id)
}

/**
 * Read is "you were shown it", and opening the list is what shows it.
 *
 * ## Why that, rather than a button
 *
 * The badge answers one question: is there anything I have not seen. A person
 * who opens the list has seen everything on it, so a badge still standing
 * afterwards is a badge that is lying, and the cure people reach for is to stop
 * looking at it. Making somebody dismiss each row individually is asking them
 * to maintain the badge by hand, on a factory floor, for a count they can
 * already read.
 *
 * The rows still draw as unread on the render that marked them, which is the
 * point: the person sees which four were new and the badge is gone.
 *
 * ## Why the ids are passed in
 *
 * Only what was actually shown is marked. The list is capped, and a second
 * screen may have added a notification between the read and this write; marking
 * "everything unread for this person" would mark that one too, and it was never
 * on anybody's screen.
 *
 * Idempotent, by `read_at IS NULL`: a re-render, a refresh and a double
 * submission all write nothing the second time, and the moment somebody first
 * saw it is not overwritten by the moment they looked again.
 */
export async function markShownRead(
  personId: string,
  ids: string[],
): Promise<number> {
  if (ids.length === 0) return 0
  const rows = await db
    .update(notifications)
    .set({ readAt: new Date() })
    .where(
      and(
        eq(notifications.personId, personId),
        inArray(notifications.id, ids),
        isNull(notifications.readAt),
      ),
    )
    .returning({ id: notifications.id })
  return rows.length
}