/**
 * Une notification écrite par un autre processus n'est pas poussée en direct,
 * ce qui est documenté. Elle doit en revanche arriver au rattrapage.
 */
import { eq, sql } from "drizzle-orm"

import { db } from "@/lib/db/client"
import { domainEvents, notifications, people, taskInstances, taskSubmissions } from "@/lib/db/schema"
import { remind } from "@/lib/services/task-commands"

const BASE = "http://127.0.0.1:4148"

const signIn = async (u: string, p: string) => {
  const r = await fetch(`${BASE}/api/auth/sign-in/username`, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ username: u, password: p }),
  })
  return r.headers.getSetCookie().map((c) => c.split(";")[0]).join("; ")
}

const read = async (cookie: string, lastId?: string, ms = 1500) => {
  const c = new AbortController()
  const res = await fetch(`${BASE}/api/notifications/stream`, {
    headers: { cookie, ...(lastId ? { "Last-Event-ID": lastId } : {}) },
    signal: c.signal,
  })
  const reader = res.body!.getReader()
  const dec = new TextDecoder()
  let text = ""
  const stop = new Promise((r) => setTimeout(r, ms))
  const pump = (async () => {
    try {
      while (true) {
        const { done, value } = await reader.read()
        if (done) break
        text += dec.decode(value)
      }
    } catch {
      /* fermé */
    }
  })()
  await stop
  c.abort()
  await pump.catch(() => {})
  return { status: res.status, type: res.headers.get("content-type"), text }
}

const cookie = await signIn("nattikan", "staff1")

// 1. Rien en attente : le flux s'ouvre et ne livre aucune notification.
const vide = await read(cookie)
console.log("ouverture :", { status: vide.status, type: vide.type })
console.log("sans rien en attente, notifications :", (vide.text.match(/event: notification/g) ?? []).length)
const ready = vide.text.match(/event: ready\ndata: (.*)/)?.[1]
console.log("curseur annoncé :", ready)

// 2. Un autre processus écrit deux notifications.
const [admin] = await db.select({ id: people.id, code: people.employeeCode }).from(people).where(eq(people.username, "narin"))
const tasks = await db
  .select({ id: taskInstances.id })
  .from(taskInstances)
  .innerJoin(people, eq(people.id, taskInstances.assigneePersonId))
  .where(sql`${people.username} = 'nattikan' and ${taskInstances.status} in ('todo','started')`)
  .limit(2)
for (const t of tasks) {
  await remind(
    { personId: admin!.id, employeeCode: admin!.code, name: "", positionTitle: null, department: null, level: "admin", tier: "admin", language: "en" },
    t.id,
    "rattrapage",
  )
}
const [ecrites] = await db.select({ n: sql<number>`count(*)::int` }).from(notifications)
console.log("écrites hors du serveur :", ecrites?.n)

// 3. Reconnexion avec le curseur : elles doivent arriver.
const reprise = await read(cookie, ready ?? undefined)
const recues = (reprise.text.match(/event: notification/g) ?? []).length
console.log("rattrapées à la reconnexion :", recues)

// 4. Une connexion neuve sans curseur ne rejoue rien.
const neuve = await read(cookie)
console.log("connexion neuve, rejouées :", (neuve.text.match(/event: notification/g) ?? []).length)

// Remise en état.
await db.delete(notifications)
await db.delete(domainEvents)
await db.delete(taskSubmissions)
const [n] = await db.select({ n: sql<number>`count(*)::int` }).from(notifications)
const st = await db.select({ s: taskInstances.status, n: sql<number>`count(*)::int` }).from(taskInstances).groupBy(taskInstances.status).orderBy(taskInstances.status)
console.log("remis :", n?.n, "notifications ;", st.map((r) => `${r.s} ${r.n}`).join(", "))

process.exit(0)