404 lines
17 KiB
TypeScript
404 lines
17 KiB
TypeScript
// 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))
|
|
},
|
|
}
|
|
}
|