L10b Betrieb & Aufräumen: Audit nach Commit, mergeCustomers atomar, Audit-Aktion read
- Aufräumpunkt h: writeAuditLog puffert innerhalb von inTransaction (AsyncLocalStorage) und schreibt nach dem Commit; bei Rollback werden die Einträge verworfen, nur „denied" bleibt. Verschachtelte Transaktionen nutzen den äußeren Puffer. - Aufräumpunkt d: mergeCustomers läuft über inTransaction (sequenziell, geschützter Statuswechsel) statt ctx.db.$transaction([...]) und ist damit auch bei RLS_ENFORCED=true atomar und in äußere Transaktionen einbettbar. - Aufräumpunkt e: AuditAction „read" (+ Label im Audit-Viewer de/en); Notdienst-Kunden- und Objektsuche protokollieren als „read" statt „export". Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
@@ -147,6 +147,7 @@
|
|||||||
"logout": "Abmeldung",
|
"logout": "Abmeldung",
|
||||||
"denied": "Abgelehnt",
|
"denied": "Abgelehnt",
|
||||||
"export": "Export",
|
"export": "Export",
|
||||||
|
"read": "Lesezugriff",
|
||||||
"import": "Import",
|
"import": "Import",
|
||||||
"provision": "Eingerichtet",
|
"provision": "Eingerichtet",
|
||||||
"approve": "Freigegeben",
|
"approve": "Freigegeben",
|
||||||
|
|||||||
@@ -147,6 +147,7 @@
|
|||||||
"logout": "Sign-out",
|
"logout": "Sign-out",
|
||||||
"denied": "Denied",
|
"denied": "Denied",
|
||||||
"export": "Export",
|
"export": "Export",
|
||||||
|
"read": "Read access",
|
||||||
"import": "Import",
|
"import": "Import",
|
||||||
"provision": "Provisioned",
|
"provision": "Provisioned",
|
||||||
"approve": "Approved",
|
"approve": "Approved",
|
||||||
|
|||||||
+76
-16
@@ -1,3 +1,4 @@
|
|||||||
|
import { AsyncLocalStorage } from "node:async_hooks";
|
||||||
import { prisma } from "./db";
|
import { prisma } from "./db";
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -6,14 +7,28 @@ import { prisma } from "./db";
|
|||||||
* filtered, and tenantId is passed explicitly by the caller.
|
* filtered, and tenantId is passed explicitly by the caller.
|
||||||
*/
|
*/
|
||||||
|
|
||||||
type AuditAction = "create" | "update" | "delete" | "login" | "logout" | "export" | "import" | "denied";
|
/** `read` (L10b): sensitive read access that must be traceable (e.g. emergency customer search). */
|
||||||
|
export type AuditAction = "create" | "update" | "delete" | "login" | "logout" | "export" | "import" | "denied" | "read";
|
||||||
|
|
||||||
|
type RequestContext = { ipAddress: string | null; userAgent: string | null };
|
||||||
|
|
||||||
|
type AuditEntry = {
|
||||||
|
tenantId: string;
|
||||||
|
actorId?: string;
|
||||||
|
action: AuditAction;
|
||||||
|
scope?: "tenant" | "platform";
|
||||||
|
entity: string;
|
||||||
|
entityId?: string;
|
||||||
|
before?: unknown;
|
||||||
|
after?: unknown;
|
||||||
|
};
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* IP address and user agent of the current request, if there is one.
|
* IP address and user agent of the current request, if there is one.
|
||||||
* Outside a request scope (workers, scripts, tests) `headers()` throws → both null.
|
* Outside a request scope (workers, scripts, tests) `headers()` throws → both null.
|
||||||
* Behind the Coolify/Traefik proxy the client IP is the first X-Forwarded-For hop.
|
* Behind the Coolify/Traefik proxy the client IP is the first X-Forwarded-For hop.
|
||||||
*/
|
*/
|
||||||
async function requestContext(): Promise<{ ipAddress: string | null; userAgent: string | null }> {
|
async function requestContext(): Promise<RequestContext> {
|
||||||
try {
|
try {
|
||||||
const { headers } = await import("next/headers");
|
const { headers } = await import("next/headers");
|
||||||
const h = await headers();
|
const h = await headers();
|
||||||
@@ -26,18 +41,8 @@ async function requestContext(): Promise<{ ipAddress: string | null; userAgent:
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
export async function writeAuditLog(entry: {
|
function insertAudit(entry: AuditEntry, request: RequestContext) {
|
||||||
tenantId: string;
|
return prisma.auditLog.create({
|
||||||
actorId?: string;
|
|
||||||
action: AuditAction;
|
|
||||||
scope?: "tenant" | "platform";
|
|
||||||
entity: string;
|
|
||||||
entityId?: string;
|
|
||||||
before?: unknown;
|
|
||||||
after?: unknown;
|
|
||||||
}) {
|
|
||||||
const ctx = await requestContext();
|
|
||||||
await prisma.auditLog.create({
|
|
||||||
data: {
|
data: {
|
||||||
tenantId: entry.tenantId,
|
tenantId: entry.tenantId,
|
||||||
scope: entry.scope ?? "tenant",
|
scope: entry.scope ?? "tenant",
|
||||||
@@ -47,12 +52,67 @@ export async function writeAuditLog(entry: {
|
|||||||
entityId: entry.entityId,
|
entityId: entry.entityId,
|
||||||
before: entry.before as object | undefined,
|
before: entry.before as object | undefined,
|
||||||
after: entry.after as object | undefined,
|
after: entry.after as object | undefined,
|
||||||
ipAddress: ctx.ipAddress,
|
ipAddress: request.ipAddress,
|
||||||
userAgent: ctx.userAgent,
|
userAgent: request.userAgent,
|
||||||
},
|
},
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* L10b (ARCHITEKTUR §4.8): audit entries written inside `inTransaction` are buffered and flushed
|
||||||
|
* after the commit. The audit insert uses the owner client (outside the tenant transaction), so
|
||||||
|
* without the buffer a rolled-back transaction would leave "create/update" entries for changes
|
||||||
|
* that never happened. On rollback only `denied` entries are kept (security relevant, independent
|
||||||
|
* of the business change).
|
||||||
|
*/
|
||||||
|
const deferredAudit = new AsyncLocalStorage<{ entries: { entry: AuditEntry; request: RequestContext }[] }>();
|
||||||
|
|
||||||
|
export async function writeAuditLog(entry: AuditEntry) {
|
||||||
|
const request = await requestContext();
|
||||||
|
const buffer = deferredAudit.getStore();
|
||||||
|
if (buffer) {
|
||||||
|
// snapshot before/after now — callers may mutate the objects after the call
|
||||||
|
buffer.entries.push({ entry: structuredCloneSafe(entry), request });
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
await insertAudit(entry, request);
|
||||||
|
}
|
||||||
|
|
||||||
|
/** Run `fn` with deferred audit writes (see above). Nested calls join the outer buffer. */
|
||||||
|
export async function withDeferredAudit<T>(fn: () => Promise<T>): Promise<T> {
|
||||||
|
if (deferredAudit.getStore()) return fn();
|
||||||
|
const buffer: { entries: { entry: AuditEntry; request: RequestContext }[] } = { entries: [] };
|
||||||
|
let result: T;
|
||||||
|
try {
|
||||||
|
result = await deferredAudit.run(buffer, fn);
|
||||||
|
} catch (err) {
|
||||||
|
await flush(buffer.entries.filter((e) => e.entry.action === "denied"));
|
||||||
|
throw err;
|
||||||
|
}
|
||||||
|
await flush(buffer.entries);
|
||||||
|
return result;
|
||||||
|
}
|
||||||
|
|
||||||
|
async function flush(entries: { entry: AuditEntry; request: RequestContext }[]) {
|
||||||
|
for (const { entry, request } of entries) {
|
||||||
|
try {
|
||||||
|
await insertAudit(entry, request);
|
||||||
|
} catch (err) {
|
||||||
|
// the business change is already committed — never turn it into an error for the caller
|
||||||
|
console.error("[audit] deferred write failed:", (err as Error).message);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/** Deep copy for JSON-like audit payloads; falls back to the original for non-cloneable values. */
|
||||||
|
function structuredCloneSafe<T>(value: T): T {
|
||||||
|
try {
|
||||||
|
return structuredClone(value);
|
||||||
|
} catch {
|
||||||
|
return value;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Audit-Eintrag der Plattform-Ebene (kein Mandantenbezug, scope="platform").
|
* Audit-Eintrag der Plattform-Ebene (kein Mandantenbezug, scope="platform").
|
||||||
* Für Superadmin-Anmeldungen und -Aktionen (Phase-1-Härtung Paket 2).
|
* Für Superadmin-Anmeldungen und -Aktionen (Phase-1-Härtung Paket 2).
|
||||||
|
|||||||
@@ -1,5 +1,6 @@
|
|||||||
import type { Session } from "next-auth";
|
import type { Session } from "next-auth";
|
||||||
import { tenantTransaction, type TenantDb } from "@/server/db";
|
import { tenantTransaction, type TenantDb } from "@/server/db";
|
||||||
|
import { withDeferredAudit } from "@/server/audit";
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Context passed to every domain service. Created by server actions (from moduleGuard)
|
* Context passed to every domain service. Created by server actions (from moduleGuard)
|
||||||
@@ -27,9 +28,11 @@ export function ctxFromGuard(g: { session: Session; db: TenantDb; permissions: R
|
|||||||
* Run a multi-step write atomically (ARCHITEKTUR §4.8). `fn` receives a ctx whose `db`
|
* Run a multi-step write atomically (ARCHITEKTUR §4.8). `fn` receives a ctx whose `db`
|
||||||
* is bound to the transaction; nested calls join the outer transaction.
|
* is bound to the transaction; nested calls join the outer transaction.
|
||||||
* Never use `ctx.db.$transaction(...)` directly — it is not atomic with RLS_ENFORCED=true.
|
* Never use `ctx.db.$transaction(...)` directly — it is not atomic with RLS_ENFORCED=true.
|
||||||
|
* Audit entries written inside `fn` are flushed after the commit and dropped on rollback
|
||||||
|
* (except `denied`), see audit.ts#withDeferredAudit.
|
||||||
*/
|
*/
|
||||||
export function inTransaction<T>(ctx: ServiceCtx, fn: (ctx: ServiceCtx) => Promise<T>): Promise<T> {
|
export function inTransaction<T>(ctx: ServiceCtx, fn: (ctx: ServiceCtx) => Promise<T>): Promise<T> {
|
||||||
return tenantTransaction(ctx.db, ctx.tenantId, (tx) => fn({ ...ctx, db: tx }));
|
return withDeferredAudit(() => tenantTransaction(ctx.db, ctx.tenantId, (tx) => fn({ ...ctx, db: tx })));
|
||||||
}
|
}
|
||||||
|
|
||||||
export function can(ctx: ServiceCtx, permission: string): boolean {
|
export function can(ctx: ServiceCtx, permission: string): boolean {
|
||||||
|
|||||||
@@ -1,5 +1,5 @@
|
|||||||
import { writeAuditLog } from "@/server/audit";
|
import { writeAuditLog } from "@/server/audit";
|
||||||
import { assertCan, ServiceError, type ServiceCtx } from "@/server/services/context";
|
import { assertCan, inTransaction, ServiceError, type ServiceCtx } from "@/server/services/context";
|
||||||
import { mergeSchema, type MergeInput } from "@/server/services/customers/schemas";
|
import { mergeSchema, type MergeInput } from "@/server/services/customers/schemas";
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -7,51 +7,58 @@ import { mergeSchema, type MergeInput } from "@/server/services/customers/schema
|
|||||||
* Contacts, sites, work orders and documents of the source are moved to the target; the source
|
* Contacts, sites, work orders and documents of the source are moved to the target; the source
|
||||||
* becomes status `merged` with `mergedIntoId`. Both records must belong to the caller's tenant
|
* becomes status `merged` with `mergedIntoId`. Both records must belong to the caller's tenant
|
||||||
* (dbForTenant) — ids of another tenant are "not found". Never triggered automatically.
|
* (dbForTenant) — ids of another tenant are "not found". Never triggered automatically.
|
||||||
|
* Runs in `inTransaction` (L10b): atomic with RLS_ENFORCED=true and joinable by callers that
|
||||||
|
* already hold a transaction (e.g. the emergency review).
|
||||||
*/
|
*/
|
||||||
export async function mergeCustomers(ctx: ServiceCtx, input: MergeInput) {
|
export async function mergeCustomers(ctx: ServiceCtx, input: MergeInput) {
|
||||||
assertCan(ctx, "customer:merge");
|
assertCan(ctx, "customer:merge");
|
||||||
const { sourceId, targetId } = mergeSchema.parse(input);
|
const { sourceId, targetId } = mergeSchema.parse(input);
|
||||||
|
|
||||||
const [source, target] = await Promise.all([
|
return inTransaction(ctx, async (tx) => {
|
||||||
ctx.db.customer.findFirst({ where: { id: sourceId, deletedAt: null } }),
|
const [source, target] = await Promise.all([
|
||||||
ctx.db.customer.findFirst({ where: { id: targetId, deletedAt: null } }),
|
tx.db.customer.findFirst({ where: { id: sourceId, deletedAt: null } }),
|
||||||
]);
|
tx.db.customer.findFirst({ where: { id: targetId, deletedAt: null } }),
|
||||||
if (!source) throw new ServiceError("not_found", "source customer not found", { field: "sourceId", reason: "not_found" });
|
]);
|
||||||
if (!target) throw new ServiceError("not_found", "target customer not found", { field: "targetId", reason: "not_found" });
|
if (!source) throw new ServiceError("not_found", "source customer not found", { field: "sourceId", reason: "not_found" });
|
||||||
if (source.status === "merged" || target.status === "merged") {
|
if (!target) throw new ServiceError("not_found", "target customer not found", { field: "targetId", reason: "not_found" });
|
||||||
throw new ServiceError("conflict", "customer already merged", { reason: "already_merged" });
|
if (source.status === "merged" || target.status === "merged") {
|
||||||
}
|
throw new ServiceError("conflict", "customer already merged", { reason: "already_merged" });
|
||||||
|
}
|
||||||
|
|
||||||
const [contacts, sites, workOrders, documents, mergedSource] = await ctx.db.$transaction([
|
// sequential on purpose: one interactive transaction, no parallel queries on the tx client
|
||||||
ctx.db.contact.updateMany({ where: { customerId: sourceId }, data: { customerId: targetId } }),
|
const contacts = await tx.db.contact.updateMany({ where: { customerId: sourceId }, data: { customerId: targetId } });
|
||||||
ctx.db.site.updateMany({ where: { customerId: sourceId }, data: { customerId: targetId } }),
|
const sites = await tx.db.site.updateMany({ where: { customerId: sourceId }, data: { customerId: targetId } });
|
||||||
// version bump: offline clients must not overwrite the re-parented order with stale data
|
// version bump: offline clients must not overwrite the re-parented order with stale data
|
||||||
ctx.db.workOrder.updateMany({ where: { customerId: sourceId }, data: { customerId: targetId, version: { increment: 1 } } }),
|
const workOrders = await tx.db.workOrder.updateMany({ where: { customerId: sourceId }, data: { customerId: targetId, version: { increment: 1 } } });
|
||||||
ctx.db.document.updateMany({ where: { customerId: sourceId }, data: { customerId: targetId } }),
|
const documents = await tx.db.document.updateMany({ where: { customerId: sourceId }, data: { customerId: targetId } });
|
||||||
ctx.db.customer.update({
|
// guarded update: a concurrent merge of the same source loses instead of merging twice
|
||||||
where: { id: sourceId },
|
const flipped = await tx.db.customer.updateMany({
|
||||||
|
where: { id: sourceId, status: { not: "merged" } },
|
||||||
data: { status: "merged", mergedIntoId: targetId, isProvisional: false },
|
data: { status: "merged", mergedIntoId: targetId, isProvisional: false },
|
||||||
}),
|
});
|
||||||
]);
|
if (flipped.count !== 1) throw new ServiceError("conflict", "customer already merged", { reason: "already_merged" });
|
||||||
|
const mergedSource = await tx.db.customer.findFirstOrThrow({ where: { id: sourceId } });
|
||||||
|
|
||||||
const moved = { contacts: contacts.count, sites: sites.count, workOrders: workOrders.count, documents: documents.count };
|
const moved = { contacts: contacts.count, sites: sites.count, workOrders: workOrders.count, documents: documents.count };
|
||||||
await writeAuditLog({
|
// deferred until commit by inTransaction (audit.ts#withDeferredAudit)
|
||||||
tenantId: ctx.tenantId,
|
await writeAuditLog({
|
||||||
actorId: ctx.userId,
|
tenantId: tx.tenantId,
|
||||||
action: "update",
|
actorId: tx.userId,
|
||||||
entity: "customer",
|
action: "update",
|
||||||
entityId: sourceId,
|
entity: "customer",
|
||||||
before: source,
|
entityId: sourceId,
|
||||||
after: { ...mergedSource, merge: { role: "source", targetId, moved } },
|
before: source,
|
||||||
|
after: { ...mergedSource, merge: { role: "source", targetId, moved } },
|
||||||
|
});
|
||||||
|
await writeAuditLog({
|
||||||
|
tenantId: tx.tenantId,
|
||||||
|
actorId: tx.userId,
|
||||||
|
action: "update",
|
||||||
|
entity: "customer",
|
||||||
|
entityId: targetId,
|
||||||
|
before: target,
|
||||||
|
after: { merge: { role: "target", sourceId, moved } },
|
||||||
|
});
|
||||||
|
return { sourceId, targetId, moved };
|
||||||
});
|
});
|
||||||
await writeAuditLog({
|
|
||||||
tenantId: ctx.tenantId,
|
|
||||||
actorId: ctx.userId,
|
|
||||||
action: "update",
|
|
||||||
entity: "customer",
|
|
||||||
entityId: targetId,
|
|
||||||
before: target,
|
|
||||||
after: { merge: { role: "target", sourceId, moved } },
|
|
||||||
});
|
|
||||||
return { sourceId, targetId, moved };
|
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -35,7 +35,7 @@ export async function searchCustomersForEmergency(ctx: ServiceCtx, rawQuery: str
|
|||||||
await writeAuditLog({
|
await writeAuditLog({
|
||||||
tenantId: ctx.tenantId,
|
tenantId: ctx.tenantId,
|
||||||
actorId: ctx.userId,
|
actorId: ctx.userId,
|
||||||
action: "export",
|
action: "read",
|
||||||
entity: "emergency_customer_search",
|
entity: "emergency_customer_search",
|
||||||
after: { query: q, resultCount: rows.length, customerIds: rows.map((r) => r.id) },
|
after: { query: q, resultCount: rows.length, customerIds: rows.map((r) => r.id) },
|
||||||
});
|
});
|
||||||
@@ -64,7 +64,7 @@ export async function listSitesForEmergency(ctx: ServiceCtx, customerId: string)
|
|||||||
await writeAuditLog({
|
await writeAuditLog({
|
||||||
tenantId: ctx.tenantId,
|
tenantId: ctx.tenantId,
|
||||||
actorId: ctx.userId,
|
actorId: ctx.userId,
|
||||||
action: "export",
|
action: "read",
|
||||||
entity: "emergency_site_lookup",
|
entity: "emergency_site_lookup",
|
||||||
entityId: customerId,
|
entityId: customerId,
|
||||||
after: { resultCount: sites.length, siteIds: sites.map((s) => s.id) },
|
after: { resultCount: sites.length, siteIds: sites.map((s) => s.id) },
|
||||||
|
|||||||
Reference in New Issue
Block a user