// The account store over kit-data: this app's own rows about a person -- the kit's users, identity
// mappings, memberships, sessions and API keys -- and the module's deletion ledger.
//
// The tables are the kit's (kit-tenancy, kit-auth); this adapter reads and changes them through the
// same repositories and scopes the kit's own use cases use, so the tenant wall holds: what crosses
// organizations is the one read the global reader is granted (which organizations a user is in), and
// every change inside an organization is made in that organization's scope. The actor of such a scope
// is the person whose account it is: an account deletion is their own act, resumed by the worker on
// their behalf.

// The entities alone: the store also runs in the Convex runtime, where kit-auth's OIDC code cannot load.
import { apiKeys, sessions } from "@conusai-kit/auth/entities"
import { AppError, type TenantScope } from "@conusai-kit/core"
import type { DataSource, SystemDataSource } from "@conusai-kit/data"
import {
  appUserIdentities,
  invitations,
  memberRemoved,
  organizations,
  orgMemberships,
  users,
  workspaceMemberships,
  workspaces,
} from "@conusai-kit/tenancy"
import { DETACHED_DISPLAY_NAME, DETACHED_ISSUER } from "../contract/constants.js"
import { accountDeletions, type DeletionStep } from "../contract/entities.js"
import { applyDeletionChange } from "../core/deletion.js"
import type { AccountMembership, AccountStore, DeletionRecord } from "../core/ports.js"

/** The most rows of one kind an account is read for at once (the kit's own `tenancy.me.get` reads 200). */
const LIMIT = 200

/**
 * The scope the identity transaction runs in. The rows it changes are global (no tenant column), so
 * the organization of the scope only sets the session's tenant variable, to an id no organization has;
 * kit-tenancy's own resolver does the same.
 */
const IDENTITY_SCOPE_ORG = "00000000-0000-0000-0000-000000000000"

const isConflict = (error: unknown): boolean => error instanceof AppError && error.status === 409

/** A delete that a concurrent run of the same step made first is done, not an error. */
async function gone(run: () => Promise<void>): Promise<void> {
  try {
    await run()
  } catch (error) {
    if (!(error instanceof AppError && error.status === 404)) throw error
  }
}

const byUser = (userId: string) => [{ field: "userId", op: "eq", value: userId }] as const

const recordOf = (row: {
  userId: string
  state: DeletionRecord["state"]
  pending: string[]
  erased: string[]
  abandoned: string[]
  attempts: number
  lastFailure: string | null
  requestedAt: string
  completedAt: string | null
  steps: DeletionStep[] | null
  provider: string | null
}): DeletionRecord => ({
  userId: row.userId,
  state: row.state,
  pending: row.pending,
  erased: row.erased,
  abandoned: row.abandoned,
  attempts: row.attempts,
  lastFailure: row.lastFailure,
  requestedAt: row.requestedAt,
  completedAt: row.completedAt,
  // A row written before the ledger recorded its steps has none: every step runs again, idempotently.
  steps: row.steps ?? [],
  provider: row.provider,
})

export function createAccountStore(deps: { data: DataSource; system: SystemDataSource }): AccountStore {
  const { data, system } = deps
  const scopeOf = (userId: string, orgId: string): TenantScope => ({ level: "org", orgId, actor: { userId } })
  const ledger = () => data.global(accountDeletions)

  const membershipRows = async (userId: string) =>
    (await system.acrossTenants(orgMemberships).list({ where: byUser(userId), limit: LIMIT })).items

  const ledgerRow = async (userId: string) => (await ledger().list({ where: byUser(userId), limit: 1 })).items[0]

  /** Every live session row of the person, page by page. */
  const liveSessions = async (userId: string) => {
    const out = []
    const where = [...byUser(userId), { field: "revokedAt", op: "isNull", isNull: true }] as const
    let cursor: string | undefined
    do {
      const page = await data.global(sessions).list({ where, limit: LIMIT, ...(cursor ? { cursor } : {}) })
      out.push(...page.items)
      cursor = page.cursor
    } while (cursor)
    return out
  }

  /**
   * The identity mapping and the profile go in one transaction. Apart, a sign-in between the two would
   * find no mapping but the user row's legacy issuer/subject columns, and the resolver's backfill would
   * map the provider identity to this row again. A concurrent change of the row (a profile sync, the
   * same step of another run) is a version conflict: read again and retry. The id stays: it is the
   * erasure reference.
   */
  const detach = async (userId: string): Promise<void> => {
    for (let attempt = 0; attempt < 8; attempt++) {
      try {
        await data.unitOfWork.run(scopeOf(userId, IDENTITY_SCOPE_ORG), async (tx) => {
          const user = await tx.global(users).get(userId)
          if (user && user.issuer !== DETACHED_ISSUER) {
            await tx.global(users).update(
              user.id,
              {
                issuer: DETACHED_ISSUER,
                subject: user.id,
                email: `${user.id}@deleted.invalid`,
                displayName: DETACHED_DISPLAY_NAME,
              },
              user.version,
            )
          }
          for (const alias of (await tx.global(appUserIdentities).list({ where: byUser(userId), limit: LIMIT }))
            .items) {
            await tx.global(appUserIdentities).delete(alias.id)
          }
        })
        return
      } catch (error) {
        if (!isConflict(error) && !(error instanceof AppError && error.status === 404)) throw error
      }
    }
    throw new AppError({ type: "/problems/conflict", title: "Conflict", status: 409, code: "conflict" })
  }

  const memberships = async (userId: string): Promise<AccountMembership[]> => {
    const out: AccountMembership[] = []
    for (const membership of await membershipRows(userId)) {
      const organization = await data.global(organizations).get(membership.orgId)
      if (!organization) continue
      const inOrg = data.forScope(scopeOf(userId, membership.orgId)).of(orgMemberships)
      const [owners, members] = await Promise.all([
        inOrg.count({ where: [{ field: "role", op: "eq", value: "owner" }] }),
        inOrg.count(),
      ])
      out.push({
        orgId: organization.id,
        slug: organization.slug,
        name: organization.name,
        role: membership.role,
        owners,
        members,
      })
    }
    return out.sort((a, b) => (a.slug < b.slug ? -1 : 1))
  }

  return {
    async emailOf(userId) {
      return (await data.global(users).get(userId))?.email ?? null
    },

    async sessionSignedInAt(userId, sessionId) {
      const row = await data.global(sessions).get(sessionId)
      // A row of another user, or one already revoked, answers nothing: it is not this person's live sign-in.
      if (!row || row.userId !== userId || row.revokedAt !== null) return null
      return row.createdAt
    },

    memberships,

    async localExport(userId) {
      const user = await data.global(users).get(userId)
      if (!user) return null
      const identities = (await data.global(appUserIdentities).list({ where: byUser(userId), limit: LIMIT })).items
      const ownSessions = (await data.global(sessions).list({ where: byUser(userId), limit: LIMIT })).items
      const organizationsOut = []
      const workspacesOut = []
      const keysOut = []
      for (const membership of await memberships(userId)) {
        const inOrg = data.forScope(scopeOf(userId, membership.orgId))
        organizationsOut.push({ slug: membership.slug, name: membership.name, role: membership.role })
        for (const row of (await inOrg.of(workspaceMemberships).list({ where: byUser(userId), limit: LIMIT })).items) {
          const workspace = await inOrg.of(workspaces).get(row.workspaceId)
          if (workspace)
            workspacesOut.push({ organization: membership.slug, workspace: workspace.slug, role: row.role })
        }
        for (const key of (await inOrg.of(apiKeys).list({ where: byUser(userId), limit: LIMIT })).items) {
          // The key's readable head and its dates; never its hash.
          keysOut.push({
            organization: membership.slug,
            name: key.name,
            prefix: key.prefix,
            createdAt: key.createdAt,
            expiresAt: key.expiresAt,
            revokedAt: key.revokedAt,
          })
        }
      }
      return {
        profile: {
          userId: user.id,
          email: user.email,
          displayName: user.displayName,
          locale: user.locale,
          createdAt: user.createdAt,
        },
        identities: identities.map((identity) => ({
          issuer: identity.issuer,
          subject: identity.subject,
          linkedAt: identity.createdAt,
        })),
        organizations: organizationsOut,
        workspaces: workspacesOut,
        // When and how long each device was signed in; never the handle or the provider's tokens.
        sessions: ownSessions.map((session) => ({
          signedInAt: session.createdAt,
          lastSeenAt: session.lastSeenAt,
          expiresAt: session.absoluteExpiresAt,
          revokedAt: session.revokedAt,
        })),
        apiKeys: keysOut,
      }
    },

    async deletion(userId) {
      const row = await ledgerRow(userId)
      return row ? recordOf(row) : null
    },

    async openDeletions(limit) {
      const page = await ledger().list({ where: [{ field: "state", op: "eq", value: "erasing" }], limit })
      return page.items.map(recordOf).sort((a, b) => (a.requestedAt < b.requestedAt ? -1 : 1))
    },

    async startDeletion({ userId, modules, provider, at }) {
      try {
        return recordOf(
          await ledger().insert({
            userId,
            state: "erasing",
            pending: [...modules],
            erased: [],
            abandoned: [],
            attempts: 0,
            lastFailure: null,
            requestedAt: at,
            completedAt: null,
            steps: [],
            provider,
          }),
        )
      } catch (error) {
        // Two requests at once: the second answers the row the first wrote (unique on userId).
        if (!isConflict(error)) throw error
        const row = await ledgerRow(userId)
        if (!row) throw error
        return recordOf(row)
      }
    },

    async changeDeletion(userId, change) {
      for (let attempt = 0; attempt < 8; attempt++) {
        const row = await ledgerRow(userId)
        if (!row)
          throw new AppError({ type: "/problems/not_found", title: "Not found", status: 404, code: "not_found" })
        const patch = applyDeletionChange(recordOf(row), change)
        if (patch === null) return recordOf(row)
        try {
          const { pending, erased, abandoned, steps, ...rest } = patch
          return recordOf(
            await ledger().update(
              row.id,
              {
                ...rest,
                ...(pending ? { pending: [...pending] } : {}),
                ...(erased ? { erased: [...erased] } : {}),
                ...(abandoned ? { abandoned: [...abandoned] } : {}),
                ...(steps ? { steps: [...steps] } : {}),
              },
              row.version,
            ),
          )
        } catch (error) {
          if (!isConflict(error)) throw error
        }
      }
      throw new AppError({ type: "/problems/conflict", title: "Conflict", status: 409, code: "conflict" })
    },

    async identities(userId) {
      const aliases = (await data.global(appUserIdentities).list({ where: byUser(userId), limit: LIMIT })).items
      const user = await data.global(users).get(userId)
      const own = user && user.issuer !== DETACHED_ISSUER ? [{ issuer: user.issuer, subject: user.subject }] : []
      // The user row's own identity is usually also an alias row: each pair once.
      const seen = new Set<string>()
      return [...aliases.map(({ issuer, subject }) => ({ issuer, subject })), ...own].filter((identity) => {
        const key = JSON.stringify([identity.issuer, identity.subject])
        if (seen.has(key)) return false
        seen.add(key)
        return true
      })
    },

    async revokeCredentials(userId, at) {
      // Every device: deleting an account ends all of its sessions, not only the one that asked. The
      // live rows are asked for by the database (revoked ones are kept for the record and could fill a
      // page), and the passes repeat until none is left: a request that slid a session's idle window
      // meanwhile makes its row's version move, and the next pass revokes it.
      for (let pass = 0; pass < 5; pass++) {
        const live = await liveSessions(userId)
        if (live.length === 0) break
        for (const row of live) {
          try {
            await data.global(sessions).update(row.id, { revokedAt: at }, row.version)
          } catch (error) {
            if (!isConflict(error)) throw error
          }
        }
      }
      if ((await liveSessions(userId)).length > 0) {
        // Still live after every pass: the step is not done, and is not recorded as done.
        throw new AppError({ type: "/problems/conflict", title: "Conflict", status: 409, code: "conflict" })
      }
      for (const membership of await membershipRows(userId)) {
        const keys = data.forScope(scopeOf(userId, membership.orgId)).of(apiKeys)
        let cursor: string | undefined
        do {
          const page = await keys.list({ where: byUser(userId), limit: LIMIT, ...(cursor ? { cursor } : {}) })
          for (const key of page.items) {
            if (key.revokedAt !== null) continue
            try {
              await keys.update(key.id, { revokedAt: at }, key.version)
            } catch (error) {
              // Revoked by a concurrent run of this step, or by an administrator: either way, revoked.
              if (!isConflict(error)) throw error
            }
          }
          cursor = page.cursor
        } while (cursor)
      }
    },

    async detachIdentity(userId) {
      const user = await data.global(users).get(userId)
      if (user && user.issuer !== DETACHED_ISSUER) {
        // The open invitations to the address, while the row still names it: one would be a way back in.
        for (const membership of await membershipRows(userId)) {
          const open = data.forScope(scopeOf(userId, membership.orgId)).of(invitations)
          for (const invitation of (
            await open.list({ where: [{ field: "email", op: "eq", value: user.email }], limit: LIMIT })
          ).items) {
            if (invitation.acceptedAt === null) await gone(() => open.delete(invitation.id))
          }
        }
      }
      await detach(userId)
    },

    async removeLocalData(userId) {
      for (const membership of await membershipRows(userId)) {
        await data.unitOfWork.run(scopeOf(userId, membership.orgId), async (tx) => {
          for (const row of (await tx.of(workspaceMemberships).list({ where: byUser(userId), limit: LIMIT })).items) {
            await gone(() => tx.of(workspaceMemberships).delete(row.id))
          }
          const own = (await tx.of(orgMemberships).list({ where: byUser(userId), limit: 1 })).items[0]
          if (!own) return
          await gone(() => tx.of(orgMemberships).delete(own.id))
          await tx.publish(memberRemoved, { orgId: membership.orgId, userId, removedBy: userId })
          if ((await tx.of(orgMemberships).count()) > 0) return
          // Nobody is left to administer it: suspended (reads only, no scheduled work), its content
          // kept for the operator, who decides what happens to an organization without members.
          for (let attempt = 0; attempt < 5; attempt++) {
            const organization = await tx.global(organizations).get(membership.orgId)
            if (!organization || organization.status === "suspended") return
            try {
              await tx.global(organizations).update(organization.id, { status: "suspended" }, organization.version)
              return
            } catch (error) {
              // Another run of this step (the request and the sweep at once) suspended it first.
              if (!isConflict(error)) throw error
            }
          }
        })
      }
      // Every session row, revoked or not, a page at a time, and any identity mapping still naming the
      // user: nothing may be left pointing at a user row that no longer exists.
      for (;;) {
        const page = await data.global(sessions).list({ where: byUser(userId), limit: LIMIT })
        if (page.items.length === 0) break
        for (const row of page.items) await gone(() => data.global(sessions).delete(row.id))
      }
      for (const alias of (await data.global(appUserIdentities).list({ where: byUser(userId), limit: LIMIT })).items) {
        await gone(() => data.global(appUserIdentities).delete(alias.id))
      }
      await gone(() => data.global(users).delete(userId))
    },
  }
}