376 lines
14 KiB
TypeScript
376 lines
14 KiB
TypeScript
// SPDX-License-Identifier: LicenseRef-ConusAI-Template-1.0
|
|
// Database transitions for the Convex file action workflows. Each handler is one real mutation;
|
|
// actions in filesActions.ts perform Blob effects between these durable state transitions.
|
|
|
|
import { type Actor, AppError, notFound, toProblem, unauthenticated, validationFailed } from "@conusai-kit/core"
|
|
import { systemActor } from "@conusai-kit/core/worker"
|
|
import { createConvexDataSource } from "@conusai-kit/data-convex"
|
|
import { executorFor } from "@conusai-kit/data-convex/runtime"
|
|
import {
|
|
beginDelete,
|
|
beginUploadIntent,
|
|
DEFAULT_FILES_POLICY,
|
|
type FileScope,
|
|
fileBeginUpload,
|
|
fileCompleteUpload,
|
|
fileDelete,
|
|
fileGet,
|
|
filePutPart,
|
|
files,
|
|
finalize,
|
|
finishDeleteRecord,
|
|
markAbandoned,
|
|
recordUploadSession,
|
|
} from "@conusai-kit/files"
|
|
import { createTenancyHooks } from "@conusai-kit/tenancy"
|
|
import type { UserIdentity } from "convex/server"
|
|
import { makeFunctionReference } from "convex/server"
|
|
import { ConvexError, type Value, v } from "convex/values"
|
|
import { internalMutation, query } from "./_generated/server"
|
|
import { CONVEX_BFF_ISSUER, convexEntities, registry, unavailableObjectStore } from "./lib/runtime"
|
|
|
|
const trustedIdentity = (identity: UserIdentity): Actor => {
|
|
if (identity.issuer !== CONVEX_BFF_ISSUER || !identity.subject) {
|
|
throwProblem(unauthenticated("file actions accept only an authenticated app BFF identity"))
|
|
}
|
|
return { userId: identity.subject }
|
|
}
|
|
|
|
function throwProblem(error: unknown): never {
|
|
if (error instanceof AppError) throw new ConvexError(toProblem(error) as unknown as Value)
|
|
throw error
|
|
}
|
|
|
|
const scopeFor = async (
|
|
data: ReturnType<typeof createConvexDataSource>,
|
|
identity: UserIdentity,
|
|
input: unknown,
|
|
permission: string,
|
|
procedure: string,
|
|
kind: "read" | "write" | "effect" = "effect",
|
|
): Promise<FileScope> => {
|
|
const actor = trustedIdentity(identity)
|
|
const tenancy = createTenancyHooks({ data, registry })
|
|
try {
|
|
const scope = await tenancy.resolveScope(
|
|
"project",
|
|
input,
|
|
actor,
|
|
(identity as UserIdentity & { conusaiOrgId?: string }).conusaiOrgId,
|
|
)
|
|
if (!(await tenancy.authorize(scope, permission, { name: procedure, kind }))) {
|
|
throw notFound(`no ${procedure} in this scope`)
|
|
}
|
|
return scope as FileScope
|
|
} catch (error) {
|
|
throwProblem(error)
|
|
}
|
|
}
|
|
|
|
const identityValidator = v.object({
|
|
issuer: v.string(),
|
|
subject: v.string(),
|
|
tokenIdentifier: v.string(),
|
|
email: v.optional(v.string()),
|
|
name: v.optional(v.string()),
|
|
conusaiOrgId: v.optional(v.string()),
|
|
})
|
|
|
|
const dataFor = (ctx: { db: unknown }) =>
|
|
createConvexDataSource({
|
|
entities: convexEntities,
|
|
executor: executorFor(ctx as unknown as Parameters<typeof executorFor>[0]),
|
|
})
|
|
const actionRef = (name: string) => makeFunctionReference<"action">(name)
|
|
|
|
export const beginIntent = internalMutation({
|
|
args: { identity: identityValidator, input: v.any() },
|
|
handler: async (ctx, { identity, input }) => {
|
|
const data = dataFor(ctx)
|
|
const parsed = fileBeginUpload.input.safeParse(input)
|
|
if (!parsed.success) throwProblem(validationFailed([{ path: "input", message: "invalid file upload input" }]))
|
|
const scope = await scopeFor(data, identity, parsed.data, "files.file.write", "files.file.beginUpload")
|
|
const intent = await beginUploadIntent({ data, policy: DEFAULT_FILES_POLICY }, scope, {
|
|
filename: parsed.data.filename,
|
|
contentType: parsed.data.contentType,
|
|
size: parsed.data.size,
|
|
uploadedBy: scope.actor.userId,
|
|
}).catch(throwProblem)
|
|
return { intent, scope: { orgId: scope.orgId, workspaceId: scope.workspaceId, projectId: scope.projectId } }
|
|
},
|
|
})
|
|
|
|
export const saveUploadSession = internalMutation({
|
|
args: { identity: identityValidator, input: v.any(), intent: v.any(), plan: v.any() },
|
|
handler: async (ctx, { identity, input, intent, plan }) => {
|
|
const data = dataFor(ctx)
|
|
const scope = await scopeFor(data, identity, input, "files.file.write", "files.file.beginUpload")
|
|
const row = await data.forScope(scope).of(files).get(intent.fileId)
|
|
if (!row || row.uploadedBy !== scope.actor.userId || row.storageKey !== intent.storageKey) {
|
|
throwProblem(notFound("no such upload"))
|
|
}
|
|
return recordUploadSession(data, scope, intent, plan).catch(throwProblem)
|
|
},
|
|
})
|
|
|
|
export const completeInfo = internalMutation({
|
|
args: { identity: identityValidator, input: v.any() },
|
|
handler: async (ctx, { identity, input }) => {
|
|
const data = dataFor(ctx)
|
|
const parsed = fileCompleteUpload.input.safeParse(input)
|
|
if (!parsed.success) throwProblem(validationFailed([{ path: "input", message: "invalid file completion input" }]))
|
|
const scope = await scopeFor(data, identity, parsed.data, "files.file.write", "files.file.completeUpload")
|
|
const row = await data.forScope(scope).of(files).get(parsed.data.fileId)
|
|
if (!row || row.uploadedBy !== scope.actor.userId) throwProblem(notFound("no such upload"))
|
|
if (row.status === "stored") {
|
|
return {
|
|
alreadyStored: true,
|
|
result: {
|
|
fileId: row.id,
|
|
filename: row.filename,
|
|
contentType: row.contentType,
|
|
size: row.size,
|
|
sha256: row.sha256,
|
|
uploadedBy: row.uploadedBy,
|
|
createdAt: row.createdAt,
|
|
storedAt: row.storedAt,
|
|
},
|
|
}
|
|
}
|
|
if (row.status !== "pending" || !row.uploadId) throwProblem(notFound("no such upload"))
|
|
return {
|
|
alreadyStored: false,
|
|
scope: { orgId: scope.orgId, workspaceId: scope.workspaceId, projectId: scope.projectId },
|
|
fileId: row.id,
|
|
storageKey: row.storageKey,
|
|
uploadId: row.uploadId,
|
|
declaredSize: row.declaredSize,
|
|
}
|
|
},
|
|
})
|
|
|
|
/** Metadata for the web BFF's streamed PUT. File bytes never become Convex action arguments. */
|
|
export const partInfo = query({
|
|
args: { input: v.any() },
|
|
handler: async (ctx, { input }) => {
|
|
const identity = await ctx.auth.getUserIdentity()
|
|
if (!identity) throwProblem(unauthenticated())
|
|
trustedIdentity(identity)
|
|
const parsed = filePutPart.input.omit({ body: true }).safeParse(input)
|
|
if (!parsed.success) throwProblem(validationFailed([{ path: "input", message: "invalid file part input" }]))
|
|
const data = dataFor(ctx)
|
|
const scope = await scopeFor(data, identity, parsed.data, "files.file.write", "files.file.putPart")
|
|
const row = await data.forScope(scope).of(files).get(parsed.data.fileId)
|
|
if (
|
|
!row ||
|
|
row.uploadedBy !== scope.actor.userId ||
|
|
row.status !== "pending" ||
|
|
!row.uploadId ||
|
|
!row.partSize ||
|
|
!row.declaredSize
|
|
) {
|
|
throwProblem(notFound("no such upload"))
|
|
}
|
|
const count = Math.ceil(row.declaredSize / row.partSize)
|
|
const part = parsed.data.part
|
|
if (!Number.isInteger(part) || part < 1 || part > count) {
|
|
throwProblem(validationFailed([{ path: "part", message: `this upload has pieces 1 to ${count}` }]))
|
|
}
|
|
return {
|
|
uploadId: row.uploadId,
|
|
storageKey: row.storageKey,
|
|
part,
|
|
size: Math.min(row.partSize, row.declaredSize - (part - 1) * row.partSize),
|
|
contentType: part === 1 ? row.contentType : undefined,
|
|
scope: { orgId: scope.orgId, workspaceId: scope.workspaceId, projectId: scope.projectId },
|
|
}
|
|
},
|
|
})
|
|
|
|
/** File metadata needed by the BFF to stream the private Blob response after authorization. */
|
|
export const downloadInfo = query({
|
|
args: { input: v.any() },
|
|
handler: async (ctx, { input }) => {
|
|
const identity = await ctx.auth.getUserIdentity()
|
|
if (!identity) throwProblem(unauthenticated())
|
|
trustedIdentity(identity)
|
|
const parsed = fileGet.input.safeParse(input)
|
|
if (!parsed.success) throwProblem(validationFailed([{ path: "input", message: "invalid file download input" }]))
|
|
const data = dataFor(ctx)
|
|
const scope = await scopeFor(data, identity, parsed.data, "files.file.read", "files.file.download", "read")
|
|
const row = await data.forScope(scope).of(files).get(parsed.data.fileId)
|
|
if (row?.status !== "stored" || row.size === null) throwProblem(notFound("no such file"))
|
|
return {
|
|
storageKey: row.storageKey,
|
|
filename: row.filename,
|
|
contentType: row.contentType,
|
|
size: row.size,
|
|
scope: { orgId: scope.orgId, workspaceId: scope.workspaceId, projectId: scope.projectId },
|
|
}
|
|
},
|
|
})
|
|
|
|
export const finalizeUpload = internalMutation({
|
|
args: { identity: identityValidator, input: v.any(), fileId: v.string(), stored: v.any() },
|
|
handler: async (ctx, { identity, input, fileId, stored }) => {
|
|
const data = dataFor(ctx)
|
|
const scope = await scopeFor(data, identity, input, "files.file.write", "files.file.completeUpload")
|
|
const row = await data.forScope(scope).of(files).get(fileId)
|
|
if (!row || row.uploadedBy !== scope.actor.userId) throwProblem(notFound("no such upload"))
|
|
return finalize({ data, store: unavailableObjectStore, policy: DEFAULT_FILES_POLICY }, scope, fileId, {
|
|
stored,
|
|
}).catch(throwProblem)
|
|
},
|
|
})
|
|
|
|
export const markDelete = internalMutation({
|
|
args: { identity: identityValidator, input: v.any() },
|
|
handler: async (ctx, { identity, input }) => {
|
|
const data = dataFor(ctx)
|
|
const parsed = fileDelete.input.safeParse(input)
|
|
if (!parsed.success) throwProblem(validationFailed([{ path: "input", message: "invalid file delete input" }]))
|
|
const scope = await scopeFor(data, identity, parsed.data, "files.file.delete", "files.file.delete", "write")
|
|
const result = await beginDelete(
|
|
{ data, store: unavailableObjectStore, policy: DEFAULT_FILES_POLICY },
|
|
scope,
|
|
parsed.data.fileId,
|
|
scope.actor.userId,
|
|
).catch(throwProblem)
|
|
return { ...result, scope: { orgId: scope.orgId, workspaceId: scope.workspaceId, projectId: scope.projectId } }
|
|
},
|
|
})
|
|
|
|
export const finalizeDelete = internalMutation({
|
|
args: { identity: identityValidator, input: v.any(), fileId: v.string() },
|
|
handler: async (ctx, { identity, input, fileId }) => {
|
|
const data = dataFor(ctx)
|
|
const scope = await scopeFor(data, identity, input, "files.file.delete", "files.file.delete", "write")
|
|
return finishDeleteRecord(data, scope, fileId)
|
|
},
|
|
})
|
|
|
|
/** Internal recovery paths do not depend on the initiating member remaining in the project. */
|
|
const systemFileScope = (ids: { orgId: string; workspaceId: string; projectId: string }): FileScope => ({
|
|
level: "project",
|
|
actor: systemActor("files.convex-recovery"),
|
|
...ids,
|
|
})
|
|
|
|
export const systemDeleteInfo = internalMutation({
|
|
args: { orgId: v.string(), workspaceId: v.string(), projectId: v.string(), fileId: v.string() },
|
|
handler: async (ctx, args) => {
|
|
const data = dataFor(ctx)
|
|
const row = await data.forScope(systemFileScope(args)).of(files).get(args.fileId)
|
|
if (row?.status !== "deleting") return null
|
|
return { storageKey: row.storageKey }
|
|
},
|
|
})
|
|
|
|
export const systemFinalizeDelete = internalMutation({
|
|
args: { orgId: v.string(), workspaceId: v.string(), projectId: v.string(), fileId: v.string() },
|
|
handler: async (ctx, args) => {
|
|
const data = dataFor(ctx)
|
|
return finishDeleteRecord(data, systemFileScope(args), args.fileId)
|
|
},
|
|
})
|
|
|
|
/** The system reconciler only claims pending uploads older than the kit's configured grace. */
|
|
export const systemPendingInfo = internalMutation({
|
|
args: { orgId: v.string(), workspaceId: v.string(), projectId: v.string(), fileId: v.string() },
|
|
handler: async (ctx, args) => {
|
|
const data = dataFor(ctx)
|
|
const row = await data.forScope(systemFileScope(args)).of(files).get(args.fileId)
|
|
const cutoff = Date.now() - DEFAULT_FILES_POLICY.pendingGraceSeconds * 1_000
|
|
if (row?.status !== "pending" || Date.parse(row.createdAt) > cutoff) {
|
|
return null
|
|
}
|
|
return { storageKey: row.storageKey, declaredSize: row.declaredSize }
|
|
},
|
|
})
|
|
|
|
export const systemAbandonUpload = internalMutation({
|
|
args: { orgId: v.string(), workspaceId: v.string(), projectId: v.string(), fileId: v.string() },
|
|
handler: async (ctx, args) => {
|
|
const data = dataFor(ctx)
|
|
return markAbandoned(
|
|
{ data, store: unavailableObjectStore, policy: DEFAULT_FILES_POLICY },
|
|
systemFileScope(args),
|
|
args.fileId,
|
|
)
|
|
},
|
|
})
|
|
|
|
export const systemFinalizeUpload = internalMutation({
|
|
args: {
|
|
orgId: v.string(),
|
|
workspaceId: v.string(),
|
|
projectId: v.string(),
|
|
fileId: v.string(),
|
|
stored: v.object({ size: v.number(), sha256: v.string() }),
|
|
},
|
|
handler: async (ctx, { stored, ...args }) => {
|
|
const data = dataFor(ctx)
|
|
return finalize(
|
|
{ data, store: unavailableObjectStore, policy: DEFAULT_FILES_POLICY },
|
|
systemFileScope(args),
|
|
args.fileId,
|
|
{ recovered: true, stored },
|
|
).catch(throwProblem)
|
|
},
|
|
})
|
|
|
|
/** Durable backstop for abandoned uploads, process crashes, and failed delete retry actions. */
|
|
export const recoverRows = internalMutation({
|
|
args: {},
|
|
handler: async (ctx) => {
|
|
const now = Date.now()
|
|
const deleting = (await ctx.db
|
|
.query("files_file")
|
|
.filter((q) => q.and(q.eq(q.field("status"), "deleting"), q.neq(q.field("deletingAt"), null)))
|
|
.order("asc")
|
|
.take(50)) as Array<Record<string, unknown>>
|
|
for (const row of deleting) {
|
|
if (
|
|
typeof row.deletingAt !== "string" ||
|
|
Date.parse(row.deletingAt) > now ||
|
|
typeof row.orgId !== "string" ||
|
|
typeof row.workspaceId !== "string" ||
|
|
typeof row.projectId !== "string" ||
|
|
typeof row.id !== "string"
|
|
)
|
|
continue
|
|
await ctx.scheduler.runAfter(0, actionRef("filesActions:cleanupDelete"), {
|
|
orgId: row.orgId as string,
|
|
workspaceId: row.workspaceId,
|
|
projectId: row.projectId,
|
|
fileId: row.id,
|
|
attempt: 0,
|
|
})
|
|
}
|
|
const staleUploads = (await ctx.db
|
|
.query("files_file")
|
|
.filter((q) => q.eq(q.field("status"), "pending"))
|
|
.order("asc")
|
|
.take(50)) as Array<Record<string, unknown>>
|
|
for (const row of staleUploads) {
|
|
if (
|
|
typeof row.createdAt !== "string" ||
|
|
Date.parse(row.createdAt) > now - DEFAULT_FILES_POLICY.pendingGraceSeconds * 1_000 ||
|
|
typeof row.orgId !== "string" ||
|
|
typeof row.workspaceId !== "string" ||
|
|
typeof row.projectId !== "string" ||
|
|
typeof row.id !== "string"
|
|
)
|
|
continue
|
|
await ctx.scheduler.runAfter(0, actionRef("filesActions:recoverUpload"), {
|
|
orgId: row.orgId as string,
|
|
workspaceId: row.workspaceId as string,
|
|
projectId: row.projectId as string,
|
|
fileId: row.id as string,
|
|
attempt: 0,
|
|
})
|
|
}
|
|
return { scheduledDeletes: deleting.length, scheduledUploads: staleUploads.length }
|
|
},
|
|
})
|