Compare commits
46
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
bcdfc536b7 | ||
|
|
77ddc0654d | ||
|
|
063206d8b0 | ||
|
|
024f4fa8c3 | ||
|
|
19df2b6a20 | ||
|
|
13e156ec7e | ||
|
|
57616d84b7 | ||
|
|
b547046963 | ||
|
|
ef84832cdb | ||
|
|
10b989351a | ||
|
|
6265734a34 | ||
|
|
f135ab2ac0 | ||
|
|
cc72e5d434 | ||
|
|
766db9e9bb | ||
|
|
5f5d31f86d | ||
|
|
4d88157cc1 | ||
|
|
8b5e23aa9b | ||
|
|
068d00febb | ||
|
|
9f74b53a4d | ||
|
|
ea16eabf79 | ||
|
|
5772b4043a | ||
|
|
afa3a888bc | ||
|
|
9c99c2c633 | ||
|
|
2ec3fe9b64 | ||
|
|
b70ae2bbfe | ||
|
|
c33d3f4d76 | ||
|
|
03e065c34d | ||
|
|
35719a6c68 | ||
|
|
140a02a797 | ||
|
|
446b9aa299 | ||
|
|
6652fde6df | ||
|
|
92eb78ba0a | ||
|
|
96d61d1308 | ||
|
|
4ea55c74bb | ||
|
|
ffd7e3341e | ||
|
|
dade37ee05 | ||
|
|
456e331d80 | ||
|
|
6a4a45a480 | ||
|
|
f29d2876a1 | ||
|
|
24f4a407c8 | ||
|
|
a0dd6b7653 | ||
|
|
8822b2adaa | ||
|
|
153c9b0cfc | ||
|
|
32182b50dd | ||
|
|
6e2d7b3068 | ||
|
|
0b50f342a3 |
@@ -280,6 +280,11 @@ OPT_CREDIT_PRICE_BOOK=
|
||||
# purchases stay disabled. Unset/`on` requires the price book + ledger.
|
||||
OPT_CREDITS_MODE=
|
||||
|
||||
# One-time trial credits granted per tenant (server-owned, idempotent, NOT a
|
||||
# payment). 0 disables. With OPT_CREDITS_MODE=on this lets reserve/settle run
|
||||
# in production before any customer purchase.
|
||||
OPT_TRIAL_CREDITS=0
|
||||
|
||||
# Central Keycloak (SIAX Portal)
|
||||
KEYCLOAK_ISSUER=https://auth.portal.siax.dev/realms/portal
|
||||
KEYCLOAK_EXPECTED_AUDIENCE=account
|
||||
|
||||
@@ -97,6 +97,9 @@ jobs:
|
||||
# cluster_name (the OpenSearch image maps dotted env keys to
|
||||
# settings).
|
||||
cluster.name: opt_ci_${{ github.run_id }}_${{ github.run_attempt }}
|
||||
# Throwaway CI cluster: never let the runner's disk fill turn indices
|
||||
# read-only (the 403 index_create_block_exception seen in run 388536).
|
||||
cluster.routing.allocation.disk.threshold_enabled: "false"
|
||||
# `--workdir` is REQUIRED (measured 2026-08-31, PR #25 CI run 7745,
|
||||
# job "integration"): this runner's act_runner (v0.6.1) merges the
|
||||
# JOB's WorkingDir (`/workspace/sax3l/0pt`) into every SERVICE
|
||||
@@ -153,6 +156,33 @@ jobs:
|
||||
if [ "$NAME" = "$WANT" ]; then FOUND="$ip"; fi
|
||||
done
|
||||
[ -n "$FOUND" ] || { echo "FATAL: no OpenSearch behind the alias identifies as $WANT"; exit 1; }
|
||||
|
||||
# The single-node test cluster inherits the runner's disk watermark.
|
||||
# On a busy runner the flood-stage watermark makes every new index
|
||||
# read-only and the suite fails with HTTP 403
|
||||
# index_create_block_exception (measured on run 388536). The container
|
||||
# is throwaway CI state, so disable the disk threshold and PROVE that
|
||||
# index creation works before the blocking tests run.
|
||||
for attempt in 1 2 3 4 5; do
|
||||
# Disabling the watermark alone does NOT clear a cluster that is
|
||||
# already read-only: the flood stage sets
|
||||
# cluster.blocks.read_only_allow_delete and every write keeps
|
||||
# returning 403. Clear that block explicitly every attempt.
|
||||
curl -sS --max-time 10 -X PUT "http://$FOUND:9200/_cluster/settings" \
|
||||
-H 'Content-Type: application/json' \
|
||||
-d '{"persistent":{"cluster.routing.allocation.disk.threshold_enabled":false,"cluster.blocks.read_only_allow_delete":null}}' >/tmp/os-settings.json || true
|
||||
cat /tmp/os-settings.json 2>/dev/null || true
|
||||
probe_status=$(curl -sS --max-time 10 -o /tmp/os-probe.json -w '%{http_code}' -X PUT "http://$FOUND:9200/opt-ci-probe-${{ github.run_id }}")
|
||||
if [ "$probe_status" = "200" ]; then
|
||||
curl -fsS --max-time 10 -X DELETE "http://$FOUND:9200/opt-ci-probe-${{ github.run_id }}" >/dev/null || true
|
||||
echo "OpenSearch accepts index creation"
|
||||
break
|
||||
fi
|
||||
echo "index creation not ready (attempt $attempt/5, HTTP $probe_status) — waiting"
|
||||
cat /tmp/os-probe.json 2>/dev/null || true
|
||||
[ "$attempt" = "5" ] && { echo "FATAL: OpenSearch still refuses index creation"; exit 1; }
|
||||
sleep 10
|
||||
done
|
||||
echo "OPENSEARCH_URL=http://$FOUND:9200" >> "$GITHUB_ENV"
|
||||
- name: Select THIS run's ClickHouse by identity
|
||||
run: |
|
||||
|
||||
@@ -0,0 +1,36 @@
|
||||
{
|
||||
"$schema": "https://git.siax.io/sax3l/siax-standard/raw/branch/main/schema/siax.acquisition-consumer.schema.json",
|
||||
"schemaVersion": "siax.acquisition-consumer/v1",
|
||||
"application": "0pt",
|
||||
"owner": "0pt",
|
||||
"tenantScoped": true,
|
||||
"capabilities": [
|
||||
"siax.acquisition.plan",
|
||||
"siax.acquisition.run",
|
||||
"siax.acquisition.status",
|
||||
"siax.acquisition.result",
|
||||
"siax.acquisition.query_existing",
|
||||
"siax.acquisition.discovery_search",
|
||||
"siax.acquisition.extract",
|
||||
"siax.acquisition.evidence"
|
||||
],
|
||||
"allowedPurposes": [
|
||||
"data.acquisition",
|
||||
"source.registration"
|
||||
],
|
||||
"dataCategories": [
|
||||
"public",
|
||||
"company",
|
||||
"product",
|
||||
"pricing",
|
||||
"vehicle",
|
||||
"property",
|
||||
"document",
|
||||
"event",
|
||||
"other"
|
||||
],
|
||||
"maxCostSEK": 25,
|
||||
"reuseExistingFacts": true,
|
||||
"mustNotOwnEngines": true,
|
||||
"engines": []
|
||||
}
|
||||
@@ -1,5 +1,6 @@
|
||||
import { NextResponse, type NextRequest } from 'next/server';
|
||||
import { PostgresCreditLedger } from '@/src/document-workbench/credit-ledger';
|
||||
import { ensureTrialCredits, readTrialCredits } from '@/src/document-workbench/trial-credits';
|
||||
import { getRepositorySql, repositoryDatabaseUrl } from '@/src/repositories/db';
|
||||
import { resolveVerifiedScope } from '@/src/http/verified-scope';
|
||||
|
||||
@@ -24,6 +25,8 @@ export async function GET(request: NextRequest) {
|
||||
|
||||
try {
|
||||
const ledger = new PostgresCreditLedger(getRepositorySql(url));
|
||||
const trial = readTrialCredits();
|
||||
if (trial > 0) await ensureTrialCredits(ledger, verified.scope.tenantId, trial);
|
||||
const balance = await ledger.balance(verified.scope.tenantId);
|
||||
return NextResponse.json({ tenant_id: verified.scope.tenantId, ...balance });
|
||||
} catch (error: unknown) {
|
||||
|
||||
@@ -27,6 +27,15 @@ export async function POST(request: NextRequest) {
|
||||
if ('response' in verified) return verified.response;
|
||||
const { tenantId } = verified.scope;
|
||||
|
||||
if (process.env.OPT_CREDITS_MODE === 'off') {
|
||||
return NextResponse.json(
|
||||
{
|
||||
error: 'CREDITS_DISABLED',
|
||||
message: 'Billing is not active yet (pilot mode). No payment can be taken.',
|
||||
},
|
||||
{ status: 503 },
|
||||
);
|
||||
}
|
||||
const book = readCreditPriceBook();
|
||||
if (!book) {
|
||||
return NextResponse.json(
|
||||
|
||||
@@ -27,6 +27,7 @@ export async function GET(request: NextRequest) {
|
||||
return NextResponse.json({
|
||||
version: book.version,
|
||||
currency: book.currency,
|
||||
purchases_enabled: process.env.OPT_CREDITS_MODE !== 'off',
|
||||
packs: book.packs.map((pack) => ({
|
||||
id: pack.id,
|
||||
credits: pack.credits,
|
||||
|
||||
@@ -3,6 +3,7 @@ import { Doc0SourceClient, Doc0SourceUnavailableError } from '@/src/document-wor
|
||||
import { Inf0OutcomeUnknownError } from '@/src/document-workbench/inf0-gateway-client';
|
||||
import { createInf0Gateway } from '@/src/document-workbench/inf0-gateway';
|
||||
import { PostgresCreditLedger } from '@/src/document-workbench/credit-ledger';
|
||||
import { ensureTrialCredits, readTrialCredits } from '@/src/document-workbench/trial-credits';
|
||||
import { creditsForUsage, readCreditPriceBook } from '@/src/document-workbench/price-book';
|
||||
import { structureDocument } from '@/src/document-workbench/structure-document';
|
||||
import { exportValidatedOutput } from '@/src/document-workbench/export-validated-output';
|
||||
@@ -163,13 +164,22 @@ export async function POST(request: NextRequest) {
|
||||
const dataClass = body.data_class ?? 'low';
|
||||
const format = body.format ?? 'json';
|
||||
|
||||
if (pipelineStage !== 'parsed') {
|
||||
return NextResponse.json(
|
||||
{
|
||||
error: 'DOCUMENT_NOT_PARSED',
|
||||
message:
|
||||
'The document has no verified text layer yet (parse_failed or not parsed). OCR is required before structuring.',
|
||||
},
|
||||
{ status: 422 },
|
||||
);
|
||||
}
|
||||
if (
|
||||
!isValidScopeId(projectId) ||
|
||||
!isValidScopeId(documentId) ||
|
||||
!isValidScopeId(versionId) ||
|
||||
typeof checksum !== 'string' ||
|
||||
!/^[0-9a-f]{64}$/.test(checksum) ||
|
||||
pipelineStage !== 'parsed' ||
|
||||
typeof instruction !== 'string' ||
|
||||
!instruction.trim() ||
|
||||
instruction.length > 4000 ||
|
||||
@@ -270,6 +280,7 @@ export async function POST(request: NextRequest) {
|
||||
creditsEnabled && databaseUrl ? new PostgresCreditLedger(getRepositorySql(databaseUrl)) : undefined;
|
||||
if (ledger) {
|
||||
try {
|
||||
await ensureTrialCredits(ledger, tenantId, readTrialCredits());
|
||||
await ledger.apply({
|
||||
tenantId,
|
||||
idempotencyKey: `reserve:${reservationId}`,
|
||||
|
||||
@@ -73,6 +73,10 @@ export default function DocumentsPage() {
|
||||
const [uploading, setUploading] = useState(false);
|
||||
const [unknownOutcome, setUnknownOutcome] = useState<unknown | null>(null);
|
||||
const inputRef = useRef<HTMLInputElement>(null);
|
||||
// One idempotency key per selected file: a retry (double click, network
|
||||
// error, reload-and-pick-the-same-file) returns the SAME document instead of
|
||||
// creating a duplicate. Picking a different file resets it.
|
||||
const uploadKey = useRef<string | null>(null);
|
||||
|
||||
const [instruction, setInstruction] = useState('');
|
||||
const [fields, setFields] = useState<FieldRow[]>([{ name: 'organization', type: 'string' }]);
|
||||
@@ -119,7 +123,9 @@ export default function DocumentsPage() {
|
||||
const form = new FormData();
|
||||
form.set('file', file);
|
||||
form.set('project_id', projectId.trim());
|
||||
form.set('idempotency_key', crypto.randomUUID());
|
||||
const key = uploadKey.current ?? crypto.randomUUID();
|
||||
uploadKey.current = key;
|
||||
form.set('idempotency_key', key);
|
||||
form.set('classification', 'restricted');
|
||||
try {
|
||||
const response = await fetch('/api/0pt/documents/upload', {
|
||||
@@ -293,7 +299,10 @@ export default function DocumentsPage() {
|
||||
ref={inputRef}
|
||||
type="file"
|
||||
className="hidden"
|
||||
onChange={(event) => setFile(event.target.files?.[0] ?? null)}
|
||||
onChange={(event) => {
|
||||
uploadKey.current = null;
|
||||
setFile(event.target.files?.[0] ?? null);
|
||||
}}
|
||||
/>
|
||||
</div>
|
||||
|
||||
|
||||
@@ -12,6 +12,7 @@ export default function CreditsPage() {
|
||||
const [balance, setBalance] = useState<{ available: number; reserved: number } | null>(null);
|
||||
const [balanceNote, setBalanceNote] = useState<string | null>(null);
|
||||
const [packs, setPacks] = useState<Pack[]>([]);
|
||||
const [purchasesEnabled, setPurchasesEnabled] = useState(true);
|
||||
const [currency, setCurrency] = useState('SEK');
|
||||
const [packsNote, setPacksNote] = useState<string | null>(null);
|
||||
const [buying, setBuying] = useState<string | null>(null);
|
||||
@@ -31,10 +32,16 @@ export default function CreditsPage() {
|
||||
}
|
||||
try {
|
||||
const r = await fetch('/api/0pt/credits/packs', { cache: 'no-store' });
|
||||
const body = (await r.json()) as { packs?: Pack[]; currency?: string; message?: string };
|
||||
const body = (await r.json()) as {
|
||||
packs?: Pack[];
|
||||
currency?: string;
|
||||
purchases_enabled?: boolean;
|
||||
message?: string;
|
||||
};
|
||||
if (r.ok && body.packs) {
|
||||
setPacks(body.packs);
|
||||
setCurrency(body.currency ?? 'SEK');
|
||||
setPurchasesEnabled(body.purchases_enabled !== false);
|
||||
} else {
|
||||
setPacksNote(
|
||||
body.message ??
|
||||
@@ -127,18 +134,30 @@ export default function CreditsPage() {
|
||||
{formatSEK(pack.amount_minor / 100)}
|
||||
</p>
|
||||
</div>
|
||||
<button
|
||||
type="button"
|
||||
onClick={() => buy(pack.id)}
|
||||
disabled={buying !== null}
|
||||
className="border border-foreground px-4 py-2 font-mono text-[11px] uppercase tracking-[0.18em] disabled:opacity-40"
|
||||
>
|
||||
{buying === pack.id ? t({ en: 'opening…', sv: 'öppnar…' }) : t({ en: 'buy', sv: 'köp' })}
|
||||
</button>
|
||||
{purchasesEnabled ? (
|
||||
<button
|
||||
type="button"
|
||||
onClick={() => buy(pack.id)}
|
||||
disabled={buying !== null}
|
||||
className="border border-foreground px-4 py-2 font-mono text-[11px] uppercase tracking-[0.18em] disabled:opacity-40"
|
||||
>
|
||||
{buying === pack.id
|
||||
? t({ en: 'opening…', sv: 'öppnar…' })
|
||||
: t({ en: 'buy', sv: 'köp' })}
|
||||
</button>
|
||||
) : null}
|
||||
</li>
|
||||
))}
|
||||
</ul>
|
||||
)}
|
||||
{!purchasesEnabled ? (
|
||||
<p className="mt-4 text-[14px] text-muted-foreground">
|
||||
{t({
|
||||
en: 'Billing is not active yet (pilot). No payment can be taken.',
|
||||
sv: 'Betalning är inte aktiverad än (pilot). Ingen betalning kan tas.',
|
||||
})}
|
||||
</p>
|
||||
) : null}
|
||||
{status ? <p className="mt-4 text-[14px] text-muted-foreground">{status}</p> : null}
|
||||
<p className="mt-4 font-mono text-[10px] uppercase tracking-[0.16em] text-muted-foreground">
|
||||
{currency}
|
||||
|
||||
@@ -37,7 +37,11 @@ is accepted by middleware as a verified principal (Bearer still wins).
|
||||
releasing it.
|
||||
6. **No price is ever read from the client.** The price book is a versioned,
|
||||
server-owned JSON document (`OPT_CREDIT_PRICE_BOOK`).
|
||||
7. **Pilot mode is explicit.** `OPT_CREDITS_MODE=off` skips reserve/settle and
|
||||
7. **Trial credits are a grant, not a payment.** `OPT_TRIAL_CREDITS=N` grants N
|
||||
credits once per tenant (`kind='grant'`, idempotency key `trial:<tenant>`),
|
||||
so the ledger is fully exercised before any real purchase. It is never read
|
||||
from a client and never looks like a Stripe payment.
|
||||
8. **Pilot mode is explicit.** `OPT_CREDITS_MODE=off` skips reserve/settle and
|
||||
reports `credits: null`. It is not a silent fallback and it does not enable
|
||||
purchases; unset/`on` requires the price book and the ledger.
|
||||
|
||||
|
||||
+5
-4
@@ -52,13 +52,15 @@
|
||||
"@radix-ui/react-toggle-group": "1.1.19",
|
||||
"@radix-ui/react-tooltip": "1.2.16",
|
||||
"@siax/0pt-decision": "workspace:*",
|
||||
"@siax/access": "workspace:*",
|
||||
"@siax/auth-zitadel": "0.1.1",
|
||||
"@siax/charts": "^0.5.0",
|
||||
"@siax/collaboration": "0.3.0",
|
||||
"@siax/config": "^0.2.1",
|
||||
"@siax/contracts": "^0.11.1",
|
||||
"@siax/contracts": "^0.12.1",
|
||||
"@siax/import-export": "^0.2.2",
|
||||
"@siax/tokens": "^0.1.2",
|
||||
"@siax/ui": "^0.15.1",
|
||||
"@siax/ui": "^0.16.13",
|
||||
"autoprefixer": "^10.4.20",
|
||||
"class-variance-authority": "^0.7.1",
|
||||
"clsx": "^2.1.1",
|
||||
@@ -81,8 +83,7 @@
|
||||
"tailwind-merge": "^3.3.1",
|
||||
"tailwindcss-animate": "^1.0.7",
|
||||
"vaul": "^1.1.2",
|
||||
"zod": "3.25.76",
|
||||
"@siax/access": "workspace:*"
|
||||
"zod": "3.25.76"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@eslint/js": "^9.15.0",
|
||||
|
||||
Generated
+56
-22
@@ -108,6 +108,9 @@ importers:
|
||||
'@siax/access':
|
||||
specifier: workspace:*
|
||||
version: link:packages/siax-access
|
||||
'@siax/auth-zitadel':
|
||||
specifier: 0.1.1
|
||||
version: 0.1.1
|
||||
'@siax/charts':
|
||||
specifier: ^0.5.0
|
||||
version: 0.5.0(react-dom@19.2.8(react@19.2.8))(react@19.2.8)
|
||||
@@ -118,8 +121,8 @@ importers:
|
||||
specifier: ^0.2.1
|
||||
version: 0.2.1(tailwindcss@4.3.3)
|
||||
'@siax/contracts':
|
||||
specifier: ^0.11.1
|
||||
version: 0.11.1
|
||||
specifier: ^0.12.1
|
||||
version: 0.12.1
|
||||
'@siax/import-export':
|
||||
specifier: ^0.2.2
|
||||
version: 0.2.2
|
||||
@@ -127,8 +130,8 @@ importers:
|
||||
specifier: ^0.1.2
|
||||
version: 0.1.2
|
||||
'@siax/ui':
|
||||
specifier: ^0.15.1
|
||||
version: 0.15.1(@types/react-dom@19.2.7(@types/react@19.2.18))(@types/react@19.2.18)(react-dom@19.2.8(react@19.2.8))(react@19.2.8)(tailwindcss@4.3.3)
|
||||
specifier: ^0.16.13
|
||||
version: 0.16.13(@types/react-dom@19.2.7(@types/react@19.2.18))(@types/react@19.2.18)(react-dom@19.2.8(react@19.2.8))(react@19.2.8)(tailwindcss@4.3.3)
|
||||
autoprefixer:
|
||||
specifier: ^10.4.20
|
||||
version: 10.5.5(postcss@8.5.28)
|
||||
@@ -1546,14 +1549,23 @@ packages:
|
||||
cpu: [x64]
|
||||
os: [win32]
|
||||
|
||||
'@siax/charts@0.5.0':
|
||||
resolution: {integrity: sha512-hQA5s0OD0L5aka9yg2mzajoW1tU+RIjZco4lrJFRw8k+EBRPES6ooKi8OAUCsuO/5u+RWrQZsea9nWJCu0oI+g==, tarball: https://git.siax.io/api/packages/sax3l/npm/%40siax%2Fcharts/-/0.5.0/charts-0.5.0.tgz}
|
||||
'@siax/analytical-context@0.5.1':
|
||||
resolution: {integrity: sha512-bqc2NGPyzLVcu4VUgz1XYsyghSM4c3ck5gBVFSzeRYMhe2cj0nf5Fm5n/sFaOuhMaKdDnX/TIR2FO6f21lgufg==, tarball: https://git.siax.io/api/packages/sax3l/npm/%40siax%2Fanalytical-context/-/0.5.1/analytical-context-0.5.1.tgz}
|
||||
|
||||
'@siax/auth-core@0.1.0':
|
||||
resolution: {integrity: sha512-oTfEO4S6pv3rQ6bSh44+fRWPiwGr9nYCf6uWSBF8g5PFUspW6nr4wKQZVAkLNwfC0Pu6o+CH8ltsuu+fY8gOQA==, tarball: https://git.siax.io/api/packages/sax3l/npm/%40siax%2Fauth-core/-/0.1.0/auth-core-0.1.0.tgz}
|
||||
|
||||
'@siax/auth-zitadel@0.1.1':
|
||||
resolution: {integrity: sha512-SmoBswG5P7szZIaETqmo1NN5RhdXLq2RDR/1wObfxHM2yUEXNSZQJznZoAb8z3lm/PvBgQhmnHiOHCfidN1ZQg==, tarball: https://git.siax.io/api/packages/sax3l/npm/%40siax%2Fauth-zitadel/-/0.1.1/auth-zitadel-0.1.1.tgz}
|
||||
|
||||
'@siax/charts@0.11.0':
|
||||
resolution: {integrity: sha512-IQnby+h6jVMaRAM7DcfUMuMbSUjd8B3+AGk9N4pXk2GDudMRMjt0bTtu75637Jc1Mf4iNonvgp8ytKlS7KdW4A==, tarball: https://git.siax.io/api/packages/sax3l/npm/%40siax%2Fcharts/-/0.11.0/charts-0.11.0.tgz}
|
||||
peerDependencies:
|
||||
react: ^19.2.0
|
||||
react-dom: ^19.2.0
|
||||
|
||||
'@siax/charts@0.5.1':
|
||||
resolution: {integrity: sha512-SFo6771jplE3fpUu2IsK11NipM5iAlM2h1N7jdVgsqqs3do+TnOi2knhIPSpUBhKfh5bZwMvO5v2TsrZmtsO+g==, tarball: https://git.siax.io/api/packages/sax3l/npm/%40siax%2Fcharts/-/0.5.1/charts-0.5.1.tgz}
|
||||
'@siax/charts@0.5.0':
|
||||
resolution: {integrity: sha512-hQA5s0OD0L5aka9yg2mzajoW1tU+RIjZco4lrJFRw8k+EBRPES6ooKi8OAUCsuO/5u+RWrQZsea9nWJCu0oI+g==, tarball: https://git.siax.io/api/packages/sax3l/npm/%40siax%2Fcharts/-/0.5.0/charts-0.5.0.tgz}
|
||||
peerDependencies:
|
||||
react: ^19.2.0
|
||||
react-dom: ^19.2.0
|
||||
@@ -1569,14 +1581,20 @@ packages:
|
||||
'@siax/contracts@0.11.1':
|
||||
resolution: {integrity: sha512-sn0ZlprJE8LcbKF4PF8wBOhdOfygOG/JbU7nB+NDFBBP/AhtvMXcxFGl1oX0GadifD7VqZrVSoCliQqY/V/HYg==, tarball: https://git.siax.io/api/packages/sax3l/npm/%40siax%2Fcontracts/-/0.11.1/contracts-0.11.1.tgz}
|
||||
|
||||
'@siax/contracts@0.12.1':
|
||||
resolution: {integrity: sha512-gSvMYRSxfbPPAqUGRjUMO1s38E4LdmIzDEpiNNIiGVminDHHwCOCybnhf2YNYaIfm6VBYoq4rP4yCERrpJregA==, tarball: https://git.siax.io/api/packages/sax3l/npm/%40siax%2Fcontracts/-/0.12.1/contracts-0.12.1.tgz}
|
||||
|
||||
'@siax/import-export@0.2.2':
|
||||
resolution: {integrity: sha512-mb2fM6lVjWh6cVdoBFOLTdyjYRtjMjlSxNtpNHZHdW51tDrNZRNZc0Mz+lVIFsJHQ79NbL2J8GsWpHmL/rVBzg==, tarball: https://git.siax.io/api/packages/sax3l/npm/%40siax%2Fimport-export/-/0.2.2/import-export-0.2.2.tgz}
|
||||
|
||||
'@siax/import-export@0.2.4':
|
||||
resolution: {integrity: sha512-xYRLF/5B7EyK3jbZCsPvDcEhZRGC/72QQpkEIsFJQqKwUxsejiQ1RabjzLoF0wNjIKlubEYnkfxDBaJMeIgKrg==, tarball: https://git.siax.io/api/packages/sax3l/npm/%40siax%2Fimport-export/-/0.2.4/import-export-0.2.4.tgz}
|
||||
|
||||
'@siax/tokens@0.1.2':
|
||||
resolution: {integrity: sha512-pz4tP8cyzhWh5gGw9GQLHjl4E/w/WemfYR5x5rdejneg+OSJYEmf6pvGCirj7247DhPqX0BH19uJK8ZN8S/Hcg==, tarball: https://git.siax.io/api/packages/sax3l/npm/%40siax%2Ftokens/-/0.1.2/tokens-0.1.2.tgz}
|
||||
|
||||
'@siax/ui@0.15.1':
|
||||
resolution: {integrity: sha512-GeU+Iwj7s3cWJUHfvt0r07dWGCGvh9jG9cS9D4EkdKb+EQGhOns2IGbDNPYRkLj/t55Qz2fjSrYCMwXO19zPDw==, tarball: https://git.siax.io/api/packages/sax3l/npm/%40siax%2Fui/-/0.15.1/ui-0.15.1.tgz}
|
||||
'@siax/ui@0.16.13':
|
||||
resolution: {integrity: sha512-qWmGexMa7kqol5O3MBDkraF/s5APAahNyGHQusDZ34YDZvGnlr3t7raWIVJMFHgz6D0Fsgyuak6210Ktn3SpHA==, tarball: https://git.siax.io/api/packages/sax3l/npm/%40siax%2Fui/-/0.16.13/ui-0.16.13.tgz}
|
||||
peerDependencies:
|
||||
react: ^19.2.0
|
||||
react-dom: ^19.2.0
|
||||
@@ -4225,16 +4243,17 @@ snapshots:
|
||||
'@rollup/rollup-win32-x64-msvc@4.62.2':
|
||||
optional: true
|
||||
|
||||
'@siax/charts@0.5.0(react-dom@19.2.8(react@19.2.8))(react@19.2.8)':
|
||||
dependencies:
|
||||
clsx: 2.1.1
|
||||
echarts: 6.1.0
|
||||
react: 19.2.8
|
||||
react-dom: 19.2.8(react@19.2.8)
|
||||
tailwind-merge: 2.6.1
|
||||
'@siax/analytical-context@0.5.1': {}
|
||||
|
||||
'@siax/charts@0.5.1(react-dom@19.2.8(react@19.2.8))(react@19.2.8)(tailwindcss@4.3.3)':
|
||||
'@siax/auth-core@0.1.0': {}
|
||||
|
||||
'@siax/auth-zitadel@0.1.1':
|
||||
dependencies:
|
||||
'@siax/auth-core': 0.1.0
|
||||
|
||||
'@siax/charts@0.11.0(react-dom@19.2.8(react@19.2.8))(react@19.2.8)(tailwindcss@4.3.3)':
|
||||
dependencies:
|
||||
'@siax/analytical-context': 0.5.1
|
||||
'@siax/config': 0.2.1(tailwindcss@4.3.3)
|
||||
clsx: 2.1.1
|
||||
echarts: 6.1.0
|
||||
@@ -4244,6 +4263,14 @@ snapshots:
|
||||
transitivePeerDependencies:
|
||||
- tailwindcss
|
||||
|
||||
'@siax/charts@0.5.0(react-dom@19.2.8(react@19.2.8))(react@19.2.8)':
|
||||
dependencies:
|
||||
clsx: 2.1.1
|
||||
echarts: 6.1.0
|
||||
react: 19.2.8
|
||||
react-dom: 19.2.8(react@19.2.8)
|
||||
tailwind-merge: 2.6.1
|
||||
|
||||
'@siax/collaboration@0.3.0': {}
|
||||
|
||||
'@siax/config@0.2.1(tailwindcss@4.3.3)':
|
||||
@@ -4252,13 +4279,19 @@ snapshots:
|
||||
|
||||
'@siax/contracts@0.11.1': {}
|
||||
|
||||
'@siax/contracts@0.12.1': {}
|
||||
|
||||
'@siax/import-export@0.2.2':
|
||||
dependencies:
|
||||
'@siax/contracts': 0.11.1
|
||||
|
||||
'@siax/import-export@0.2.4':
|
||||
dependencies:
|
||||
'@siax/contracts': 0.12.1
|
||||
|
||||
'@siax/tokens@0.1.2': {}
|
||||
|
||||
'@siax/ui@0.15.1(@types/react-dom@19.2.7(@types/react@19.2.18))(@types/react@19.2.18)(react-dom@19.2.8(react@19.2.8))(react@19.2.8)(tailwindcss@4.3.3)':
|
||||
'@siax/ui@0.16.13(@types/react-dom@19.2.7(@types/react@19.2.18))(@types/react@19.2.18)(react-dom@19.2.8(react@19.2.8))(react@19.2.8)(tailwindcss@4.3.3)':
|
||||
dependencies:
|
||||
'@radix-ui/react-alert-dialog': 1.1.23(@types/react-dom@19.2.7(@types/react@19.2.18))(@types/react@19.2.18)(react-dom@19.2.8(react@19.2.8))(react@19.2.8)
|
||||
'@radix-ui/react-aspect-ratio': 1.1.15(@types/react-dom@19.2.7(@types/react@19.2.18))(@types/react@19.2.18)(react-dom@19.2.8(react@19.2.8))(react@19.2.8)
|
||||
@@ -4281,10 +4314,11 @@ snapshots:
|
||||
'@radix-ui/react-toggle': 1.1.18(@types/react-dom@19.2.7(@types/react@19.2.18))(@types/react@19.2.18)(react-dom@19.2.8(react@19.2.8))(react@19.2.8)
|
||||
'@radix-ui/react-toggle-group': 1.1.19(@types/react-dom@19.2.7(@types/react@19.2.18))(@types/react@19.2.18)(react-dom@19.2.8(react@19.2.8))(react@19.2.8)
|
||||
'@radix-ui/react-tooltip': 1.2.16(@types/react-dom@19.2.7(@types/react@19.2.18))(@types/react@19.2.18)(react-dom@19.2.8(react@19.2.8))(react@19.2.8)
|
||||
'@siax/charts': 0.5.1(react-dom@19.2.8(react@19.2.8))(react@19.2.8)(tailwindcss@4.3.3)
|
||||
'@siax/analytical-context': 0.5.1
|
||||
'@siax/charts': 0.11.0(react-dom@19.2.8(react@19.2.8))(react@19.2.8)(tailwindcss@4.3.3)
|
||||
'@siax/config': 0.2.1(tailwindcss@4.3.3)
|
||||
'@siax/contracts': 0.11.1
|
||||
'@siax/import-export': 0.2.2
|
||||
'@siax/contracts': 0.12.1
|
||||
'@siax/import-export': 0.2.4
|
||||
'@siax/tokens': 0.1.2
|
||||
'@tanstack/react-table': 8.21.3(react-dom@19.2.8(react@19.2.8))(react@19.2.8)
|
||||
'@tanstack/react-virtual': 3.14.13(react-dom@19.2.8(react@19.2.8))(react@19.2.8)
|
||||
|
||||
@@ -0,0 +1,10 @@
|
||||
FROM python:3.12-slim
|
||||
WORKDIR /app
|
||||
ENV PYTHONUNBUFFERED=1 PIP_NO_CACHE_DIR=1
|
||||
COPY requirements.txt .
|
||||
RUN pip install -r requirements.txt
|
||||
COPY app.py ingest_aud0.py ./
|
||||
EXPOSE 8096
|
||||
HEALTHCHECK --interval=30s --timeout=5s --retries=3 \
|
||||
CMD python -c "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8096/health', timeout=3)"
|
||||
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8096"]
|
||||
@@ -0,0 +1,47 @@
|
||||
# 0pt analytics HTTP surface
|
||||
|
||||
The missing read plane between API0's `/analytics/*` contract and the
|
||||
`opt_analytics` ClickHouse database.
|
||||
|
||||
## Surface
|
||||
|
||||
Serves the BI-P068 operation paths under `/api/0pt/analytics`:
|
||||
|
||||
- `GET /health`
|
||||
- `GET /datasets`, `GET /datasets/{id}`
|
||||
- `GET /metrics`, `GET /metrics/{id}`, `GET /metrics/{id}/explain`
|
||||
- `POST /metrics/{id}/query`
|
||||
- `POST /query`, `POST /query/estimate`, `POST /query/explain`
|
||||
- `GET /query/{id}` (404: executions are not persisted yet)
|
||||
- `GET /dashboards`, `GET /reports` (empty until the BI metadata lane writes them)
|
||||
- `GET /search`
|
||||
|
||||
Identity: the caller's Keycloak token is verified against the portal realm
|
||||
JWKS; `tenant_id` is taken from the token and cross-checked against
|
||||
`X-Tenant-Id`. An `X-Internal-Token` path exists only for the ingestion worker.
|
||||
|
||||
## Ingest
|
||||
|
||||
`ingest_aud0.py` maps AUD0 audit events to `opt_analytics_event`
|
||||
(`dataset_id=aud0.events`, `metric=event_count`, dimensions event_type/app_id/
|
||||
capability_id/risk_level). It is idempotent per `event_id`.
|
||||
|
||||
```bash
|
||||
AUD0_TOKEN=... AUD0_TENANT_ID=... CLICKHOUSE_URL=http://opt-analytics-clickhouse:8123 \
|
||||
python ingest_aud0.py --limit 50000
|
||||
```
|
||||
|
||||
## Run
|
||||
|
||||
```bash
|
||||
pip install -r requirements.txt
|
||||
CLICKHOUSE_URL=http://opt-analytics-clickhouse:8123 \
|
||||
KEYCLOAK_ISSUER=https://auth.portal.siax.dev/realms/portal \
|
||||
uvicorn app:app --host 0.0.0.0 --port 8096
|
||||
```
|
||||
|
||||
## Test
|
||||
|
||||
```bash
|
||||
python -m pytest tests -q
|
||||
```
|
||||
@@ -0,0 +1,464 @@
|
||||
"""
|
||||
0pt analytics HTTP surface (BI-P068-compatible).
|
||||
|
||||
This is the missing read plane between API0's `/analytics/*` contract and the
|
||||
`opt_analytics` ClickHouse database. It does not invent a second analytics
|
||||
model:
|
||||
|
||||
* datasets/metrics are the distinct ids actually present in
|
||||
`opt_analytics_event` / `opt_metric_daily` / `opt_dataset_projection`;
|
||||
* query/estimate/explain are computed against the tenant-forced ClickHouse
|
||||
rows, never against a local cache;
|
||||
* dashboards/reports are honest empty lists until the 0pt BI metadata lane
|
||||
writes them;
|
||||
* every response is either real ClickHouse data or an explicit error.
|
||||
|
||||
IDENTITY. API0 verifies the caller and forwards the caller's own Keycloak
|
||||
token plus `X-Tenant-Id`. This service verifies the same token against the
|
||||
portal realm's JWKS and cross-checks `tenant_id` against `X-Tenant-Id`. A
|
||||
service-to-service `X-Internal-Token` is also accepted for the ingestion
|
||||
worker, but never for a request that reads tenant data without an explicit,
|
||||
already-verified `X-Tenant-Id`.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import time
|
||||
from functools import lru_cache
|
||||
from typing import Any, Iterable, Literal
|
||||
|
||||
import httpx
|
||||
import jwt
|
||||
from fastapi import FastAPI, Header, HTTPException, Query, Request
|
||||
from fastapi.responses import JSONResponse
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
CLICKHOUSE_URL = os.getenv("CLICKHOUSE_URL", "").rstrip("/")
|
||||
CLICKHOUSE_USER = os.getenv("CLICKHOUSE_USER", "default")
|
||||
CLICKHOUSE_PASSWORD = os.getenv("CLICKHOUSE_PASSWORD", "")
|
||||
CLICKHOUSE_DATABASE = os.getenv("CLICKHOUSE_DATABASE", "opt_analytics")
|
||||
KEYCLOAK_ISSUER = os.getenv("KEYCLOAK_ISSUER", "").rstrip("/")
|
||||
KEYCLOAK_AUDIENCE = os.getenv("KEYCLOAK_AUDIENCE", "account")
|
||||
KEYCLOAK_AUTHORIZED_PARTY = os.getenv("KEYCLOAK_AUTHORIZED_PARTY", "siax-portal")
|
||||
INTERNAL_TOKEN = os.getenv("INTERNAL_TOKEN", "")
|
||||
ALLOWED_DIMENSIONS = {"event_type", "app_id", "capability_id", "risk_level", "source_service"}
|
||||
|
||||
app = FastAPI(title="0pt analytics", version="0.1")
|
||||
|
||||
|
||||
class AnalyticsError(Exception):
|
||||
def __init__(self, code: str, message: str, status: int = 400):
|
||||
self.code = code
|
||||
self.message = message
|
||||
self.status = status
|
||||
super().__init__(message)
|
||||
|
||||
|
||||
@app.exception_handler(AnalyticsError)
|
||||
async def analytics_error_handler(_request: Request, exc: AnalyticsError) -> JSONResponse:
|
||||
return JSONResponse(status_code=exc.status, content={"error": {"code": exc.code, "message": exc.message}})
|
||||
|
||||
|
||||
# ── identity ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@lru_cache(maxsize=1)
|
||||
def _jwks_client() -> jwt.PyJWKClient:
|
||||
if not KEYCLOAK_ISSUER:
|
||||
raise AnalyticsError("ANALYTICS_NOT_CONFIGURED", "KEYCLOAK_ISSUER is not set", 503)
|
||||
return jwt.PyJWKClient(f"{KEYCLOAK_ISSUER}/protocol/openid-connect/certs", cache_keys=True, lifespan=3600)
|
||||
|
||||
|
||||
def _bearer(request: Request) -> str | None:
|
||||
raw = request.headers.get("authorization", "")
|
||||
if not raw.lower().startswith("bearer "):
|
||||
return None
|
||||
return raw[7:].strip() or None
|
||||
|
||||
|
||||
def verified_identity(request: Request, x_tenant_id: str | None) -> str:
|
||||
"""Return the tenant the request is allowed to read.
|
||||
|
||||
An `X-Internal-Token` may only be used by the ingestion worker and never
|
||||
substitutes for a tenant claim. For user/API0 calls the JWT is the source
|
||||
of truth; a mismatching X-Tenant-Id is a hard 403, not a preference.
|
||||
"""
|
||||
if INTERNAL_TOKEN and request.headers.get("x-internal-token") == INTERNAL_TOKEN:
|
||||
if not x_tenant_id:
|
||||
raise AnalyticsError("TENANT_REQUIRED", "X-Tenant-Id is required for internal calls", 400)
|
||||
return x_tenant_id
|
||||
|
||||
token = _bearer(request)
|
||||
if not token:
|
||||
raise AnalyticsError("UNAUTHENTICATED", "A bearer token is required", 401)
|
||||
try:
|
||||
signing_key = _jwks_client().get_signing_key_from_jwt(token)
|
||||
claims = jwt.decode(
|
||||
token,
|
||||
signing_key.key,
|
||||
algorithms=["RS256", "RS384", "RS512"],
|
||||
audience=KEYCLOAK_AUDIENCE,
|
||||
issuer=KEYCLOAK_ISSUER,
|
||||
options={"require": ["exp", "iss", "sub"]},
|
||||
)
|
||||
except jwt.PyJWTError as exc:
|
||||
raise AnalyticsError("UNAUTHENTICATED", f"Invalid bearer token: {exc}", 401) from exc
|
||||
|
||||
if KEYCLOAK_AUTHORIZED_PARTY and claims.get("azp") not in (None, KEYCLOAK_AUTHORIZED_PARTY):
|
||||
raise AnalyticsError("FORBIDDEN", "Token was issued for a different client", 403)
|
||||
tenant = claims.get("tenant_id")
|
||||
if not isinstance(tenant, str) or not tenant.strip():
|
||||
raise AnalyticsError("NO_TENANT", "Token is not bound to a tenant", 403)
|
||||
if x_tenant_id and x_tenant_id != tenant:
|
||||
raise AnalyticsError("TENANT_MISMATCH", "X-Tenant-Id does not match the verified tenant", 403)
|
||||
return tenant
|
||||
|
||||
|
||||
# ── clickhouse ──────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _ch_client() -> httpx.Client:
|
||||
if not CLICKHOUSE_URL:
|
||||
raise AnalyticsError("ANALYTICS_NOT_CONFIGURED", "CLICKHOUSE_URL is not set", 503)
|
||||
auth = (CLICKHOUSE_USER, CLICKHOUSE_PASSWORD) if CLICKHOUSE_PASSWORD else None
|
||||
return httpx.Client(base_url=CLICKHOUSE_URL, auth=auth, timeout=20.0)
|
||||
|
||||
|
||||
def ch_json(sql: str, params: dict[str, str] | None = None) -> dict[str, Any]:
|
||||
params = params or {}
|
||||
query_params = {
|
||||
"database": CLICKHOUSE_DATABASE,
|
||||
"output_format_json_quote_64bit_integers": "0",
|
||||
**{f"param_{key}": value for key, value in params.items()},
|
||||
}
|
||||
try:
|
||||
with _ch_client() as client:
|
||||
response = client.post("/", params=query_params, content=sql)
|
||||
except httpx.HTTPError as exc:
|
||||
raise AnalyticsError("CLICKHOUSE_UNREACHABLE", str(exc), 503) from exc
|
||||
if response.status_code != 200:
|
||||
raise AnalyticsError("CLICKHOUSE_REFUSED", response.text[:500], 502)
|
||||
try:
|
||||
return response.json()
|
||||
except ValueError as exc:
|
||||
raise AnalyticsError("CLICKHOUSE_BAD_RESPONSE", "ClickHouse did not return JSON", 502) from exc
|
||||
|
||||
|
||||
def _literal(value: str) -> str:
|
||||
# ClickHouse query parameters are preferable; the only values interpolated
|
||||
# here are identifiers from ALLOWED_DIMENSIONS, never caller text.
|
||||
if value not in ALLOWED_DIMENSIONS:
|
||||
raise AnalyticsError("INVALID_DIMENSION", f"Unsupported dimension: {value}")
|
||||
return value
|
||||
|
||||
|
||||
def rows(sql: str, params: dict[str, str] | None = None) -> list[dict[str, Any]]:
|
||||
return ch_json(sql, params).get("data", [])
|
||||
|
||||
|
||||
def scalar(sql: str, params: dict[str, str] | None = None) -> Any:
|
||||
data = rows(sql, params)
|
||||
if not data:
|
||||
return None
|
||||
return next(iter(data[0].values()))
|
||||
|
||||
|
||||
def build_query(
|
||||
*,
|
||||
tenant: str,
|
||||
dataset: str | None,
|
||||
metrics: Iterable[str],
|
||||
dimensions: Iterable[str],
|
||||
filters: dict[str, str] | None,
|
||||
from_time: str | None,
|
||||
to_time: str | None,
|
||||
limit: int,
|
||||
) -> tuple[str, dict[str, str]]:
|
||||
dims = [_literal(d) for d in dimensions]
|
||||
metric_list = [m for m in metrics if m.strip()]
|
||||
if not metric_list:
|
||||
raise AnalyticsError("INVALID_REQUEST", "At least one metric is required")
|
||||
where = ["tenant_id = {tenant:String}"]
|
||||
params: dict[str, str] = {"tenant": tenant, "limit": str(max(1, min(limit, 1000)))}
|
||||
if dataset:
|
||||
where.append("dataset_id = {dataset:String}")
|
||||
params["dataset"] = dataset
|
||||
metric_params = []
|
||||
for i, metric in enumerate(metric_list):
|
||||
key = f"metric{i}"
|
||||
metric_params.append(f"{{{key}:String}}")
|
||||
params[key] = metric
|
||||
where.append(f"metric IN ({', '.join(metric_params)})")
|
||||
for i, (key, value) in enumerate((filters or {}).items()):
|
||||
dim = _literal(key)
|
||||
p = f"filter{i}"
|
||||
where.append(f"dim_values[indexOf(dim_names, '{dim}')] = {{{p}:String}}")
|
||||
params[p] = value
|
||||
if from_time:
|
||||
where.append("event_time >= {from:String}")
|
||||
params["from"] = from_time
|
||||
if to_time:
|
||||
where.append("event_time <= {to:String}")
|
||||
params["to"] = to_time
|
||||
|
||||
select_dims = []
|
||||
for i, dim in enumerate(dims):
|
||||
select_dims.append(f"dim_values[indexOf(dim_names, '{dim}')] AS d{i}")
|
||||
select = ", ".join(select_dims + ["count() AS value"])
|
||||
group = f" GROUP BY {', '.join(f'd{i}' for i in range(len(dims)))}" if dims else ""
|
||||
sql = (
|
||||
f"SELECT {select} FROM {CLICKHOUSE_DATABASE}.opt_analytics_event "
|
||||
f"WHERE {' AND '.join(where)}{group} ORDER BY value DESC LIMIT {{limit:UInt32}} FORMAT JSON"
|
||||
)
|
||||
return sql, params
|
||||
|
||||
|
||||
# ── models ─────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class QueryRequest(BaseModel):
|
||||
dataset: str | None = None
|
||||
metrics: list[str] = Field(default_factory=list)
|
||||
dimensions: list[str] = Field(default_factory=list)
|
||||
filters: dict[str, str] | None = None
|
||||
timeRange: dict[str, str] | None = None
|
||||
limit: int = 100
|
||||
|
||||
|
||||
# ── surface ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@app.get("/health")
|
||||
def health() -> dict[str, Any]:
|
||||
try:
|
||||
ch_json("SELECT 1")
|
||||
return {"status": "ok", "providers": {"clickhouse": {"status": "ok"}}}
|
||||
except AnalyticsError as exc:
|
||||
return {"status": "error", "providers": {"clickhouse": {"status": "error", "detail": exc.message}}}
|
||||
|
||||
|
||||
@app.get("/api/0pt/analytics/health")
|
||||
def analytics_health() -> dict[str, Any]:
|
||||
return health()
|
||||
|
||||
|
||||
@app.get("/api/0pt/analytics/datasets")
|
||||
def list_datasets(request: Request, x_tenant_id: str | None = Header(default=None)) -> list[dict[str, Any]]:
|
||||
tenant = verified_identity(request, x_tenant_id)
|
||||
rows_ = rows(
|
||||
f"SELECT dataset_id, count() AS events FROM {CLICKHOUSE_DATABASE}.opt_analytics_event "
|
||||
"WHERE tenant_id = {tenant:String} GROUP BY dataset_id ORDER BY dataset_id FORMAT JSON",
|
||||
{"tenant": tenant},
|
||||
)
|
||||
if not rows_:
|
||||
rows_ = rows(
|
||||
f"SELECT dataset_id, count() AS rows FROM {CLICKHOUSE_DATABASE}.opt_dataset_projection "
|
||||
"WHERE tenant_id = {tenant:String} GROUP BY dataset_id ORDER BY dataset_id FORMAT JSON",
|
||||
{"tenant": tenant},
|
||||
)
|
||||
return [
|
||||
{
|
||||
"id": r["dataset_id"],
|
||||
"name": r["dataset_id"],
|
||||
"tier": "SERVING",
|
||||
"owner": "0pt",
|
||||
"source": "0pt analytics",
|
||||
"tenant_id": tenant,
|
||||
**({"events": r.get("events")} if "events" in r else {}),
|
||||
**({"rows": r.get("rows")} if "rows" in r else {}),
|
||||
}
|
||||
for r in rows_
|
||||
]
|
||||
|
||||
|
||||
@app.get("/api/0pt/analytics/datasets/{dataset_id}")
|
||||
def get_dataset(dataset_id: str, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]:
|
||||
tenant = verified_identity(request, x_tenant_id)
|
||||
found = [d for d in list_datasets(request, x_tenant_id) if d["id"] == dataset_id]
|
||||
if not found:
|
||||
raise AnalyticsError("NOT_FOUND", f"dataset {dataset_id} does not exist for this tenant", 404)
|
||||
return found[0]
|
||||
|
||||
|
||||
@app.get("/api/0pt/analytics/metrics")
|
||||
def list_metrics(request: Request, x_tenant_id: str | None = Header(default=None)) -> list[dict[str, Any]]:
|
||||
tenant = verified_identity(request, x_tenant_id)
|
||||
rows_ = rows(
|
||||
f"SELECT metric, count() AS events FROM {CLICKHOUSE_DATABASE}.opt_analytics_event "
|
||||
"WHERE tenant_id = {tenant:String} GROUP BY metric ORDER BY metric FORMAT JSON",
|
||||
{"tenant": tenant},
|
||||
)
|
||||
if not rows_:
|
||||
rows_ = rows(
|
||||
f"SELECT metric, sum(count_state) AS events FROM {CLICKHOUSE_DATABASE}.opt_metric_daily "
|
||||
"WHERE tenant_id = {tenant:String} GROUP BY metric ORDER BY metric FORMAT JSON",
|
||||
{"tenant": tenant},
|
||||
)
|
||||
dims = rows(
|
||||
f"SELECT DISTINCT arrayJoin(dim_names) AS name FROM {CLICKHOUSE_DATABASE}.opt_analytics_event "
|
||||
"WHERE tenant_id = {tenant:String} FORMAT JSON",
|
||||
{"tenant": tenant},
|
||||
)
|
||||
dimension_defs = [{"name": d["name"], "type": "string"} for d in dims]
|
||||
return [
|
||||
{
|
||||
"id": r["metric"],
|
||||
"name": r["metric"],
|
||||
"description": f"0pt analytics metric {r['metric']}",
|
||||
"datasets": ["aud0.events"],
|
||||
"dimensions": dimension_defs,
|
||||
"tenant_id": tenant,
|
||||
**({"events": r.get("events")} if "events" in r else {}),
|
||||
}
|
||||
for r in rows_
|
||||
]
|
||||
|
||||
|
||||
@app.get("/api/0pt/analytics/metrics/{metric_id}")
|
||||
def get_metric(metric_id: str, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]:
|
||||
tenant = verified_identity(request, x_tenant_id)
|
||||
found = [m for m in list_metrics(request, x_tenant_id) if m["id"] == metric_id]
|
||||
if not found:
|
||||
raise AnalyticsError("NOT_FOUND", f"metric {metric_id} does not exist for this tenant", 404)
|
||||
return found[0]
|
||||
|
||||
|
||||
def _run_query(request: Request, x_tenant_id: str | None, body: QueryRequest, metric: str | None = None) -> dict[str, Any]:
|
||||
tenant = verified_identity(request, x_tenant_id)
|
||||
metrics = [metric] if metric else body.metrics
|
||||
dataset = body.dataset or "aud0.events"
|
||||
sql, params = build_query(
|
||||
tenant=tenant,
|
||||
dataset=dataset,
|
||||
metrics=metrics,
|
||||
dimensions=body.dimensions,
|
||||
filters=body.filters,
|
||||
from_time=(body.timeRange or {}).get("from"),
|
||||
to_time=(body.timeRange or {}).get("to"),
|
||||
limit=body.limit,
|
||||
)
|
||||
started = time.time()
|
||||
data = ch_json(sql, params)
|
||||
result_rows = data.get("data", [])
|
||||
duration_ms = int((time.time() - started) * 1000)
|
||||
columns = []
|
||||
if result_rows:
|
||||
columns = [{"name": key, "type": "string" if key.startswith("d") else "number"} for key in result_rows[0].keys()]
|
||||
return {
|
||||
"query_id": f"ch-{int(time.time() * 1000)}",
|
||||
"status": "succeeded",
|
||||
"engine": "clickhouse",
|
||||
"submitted_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
||||
"completed_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
||||
"duration": duration_ms,
|
||||
"rows": len(result_rows),
|
||||
"tenant_id": tenant,
|
||||
"result": {"columns": columns, "rows": result_rows, "truncated": len(result_rows) >= body.limit},
|
||||
}
|
||||
|
||||
|
||||
@app.post("/api/0pt/analytics/query")
|
||||
def execute_query(body: QueryRequest, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]:
|
||||
return _run_query(request, x_tenant_id, body)
|
||||
|
||||
|
||||
@app.post("/api/0pt/analytics/metrics/{metric_id}/query")
|
||||
def execute_metric(metric_id: str, body: QueryRequest, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]:
|
||||
return _run_query(request, x_tenant_id, body, metric=metric_id)
|
||||
|
||||
|
||||
@app.post("/api/0pt/analytics/query/estimate")
|
||||
def estimate_query(body: QueryRequest, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]:
|
||||
tenant = verified_identity(request, x_tenant_id)
|
||||
sql, params = build_query(
|
||||
tenant=tenant,
|
||||
dataset=body.dataset or "aud0.events",
|
||||
metrics=body.metrics or ["event_count"],
|
||||
dimensions=body.dimensions,
|
||||
filters=body.filters,
|
||||
from_time=(body.timeRange or {}).get("from"),
|
||||
to_time=(body.timeRange or {}).get("to"),
|
||||
limit=body.limit,
|
||||
)
|
||||
count = scalar(
|
||||
f"SELECT count() FROM {CLICKHOUSE_DATABASE}.opt_analytics_event WHERE tenant_id = {{tenant:String}} FORMAT JSON",
|
||||
{"tenant": tenant},
|
||||
)
|
||||
return {"estimated_rows_to_read": count or 0, "within_budget": True, "rows": count or 0}
|
||||
|
||||
|
||||
@app.post("/api/0pt/analytics/query/explain")
|
||||
def explain_query(body: QueryRequest, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]:
|
||||
tenant = verified_identity(request, x_tenant_id)
|
||||
sql, _params = build_query(
|
||||
tenant=tenant,
|
||||
dataset=body.dataset or "aud0.events",
|
||||
metrics=body.metrics or ["event_count"],
|
||||
dimensions=body.dimensions,
|
||||
filters=body.filters,
|
||||
from_time=(body.timeRange or {}).get("from"),
|
||||
to_time=(body.timeRange or {}).get("to"),
|
||||
limit=body.limit,
|
||||
)
|
||||
return {"plan": sql, "forced_tenant_predicate": "tenant_id = {tenant:String}", "table": "opt_analytics_event"}
|
||||
|
||||
|
||||
@app.get("/api/0pt/analytics/metrics/{metric_id}/explain")
|
||||
def explain_metric(metric_id: str, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]:
|
||||
return explain_query(QueryRequest(metrics=[metric_id]), request, x_tenant_id)
|
||||
|
||||
|
||||
@app.get("/api/0pt/analytics/dashboards")
|
||||
def list_dashboards(request: Request, x_tenant_id: str | None = Header(default=None)) -> list[dict[str, Any]]:
|
||||
verified_identity(request, x_tenant_id)
|
||||
return []
|
||||
|
||||
|
||||
@app.get("/api/0pt/analytics/dashboards/{dashboard_id}")
|
||||
def get_dashboard(dashboard_id: str, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]:
|
||||
verified_identity(request, x_tenant_id)
|
||||
raise AnalyticsError("NOT_FOUND", f"dashboard {dashboard_id} does not exist", 404)
|
||||
|
||||
|
||||
@app.get("/api/0pt/analytics/reports")
|
||||
def list_reports(request: Request, x_tenant_id: str | None = Header(default=None)) -> list[dict[str, Any]]:
|
||||
verified_identity(request, x_tenant_id)
|
||||
return []
|
||||
|
||||
|
||||
@app.get("/api/0pt/analytics/reports/{report_id}")
|
||||
def get_report(report_id: str, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]:
|
||||
verified_identity(request, x_tenant_id)
|
||||
raise AnalyticsError("NOT_FOUND", f"report {report_id} does not exist", 404)
|
||||
|
||||
|
||||
@app.get("/api/0pt/analytics/search")
|
||||
def search_catalog(
|
||||
request: Request,
|
||||
q: str = Query(default=""),
|
||||
types: str = Query(default=""),
|
||||
limit: int = Query(default=20, ge=1, le=100),
|
||||
x_tenant_id: str | None = Header(default=None),
|
||||
) -> dict[str, Any]:
|
||||
tenant = verified_identity(request, x_tenant_id)
|
||||
wanted = {t.strip() for t in types.split(",") if t.strip()}
|
||||
hits: list[dict[str, Any]] = []
|
||||
for dataset in list_datasets(request, x_tenant_id):
|
||||
if (not wanted or "dataset" in wanted) and (not q or q.lower() in dataset["id"].lower()):
|
||||
hits.append({**dataset, "type": "dataset", "canonical": False, "canonicalOperation": "getDataset"})
|
||||
for metric in list_metrics(request, x_tenant_id):
|
||||
if (not wanted or "metric" in wanted) and (not q or q.lower() in metric["id"].lower()):
|
||||
hits.append({**metric, "type": "metric", "canonical": False, "canonicalOperation": "getMetric"})
|
||||
return {"query": q, "derived": True, "hits": hits[:limit], "tenant_id": tenant}
|
||||
|
||||
|
||||
@app.get("/api/0pt/analytics/query/{query_id}")
|
||||
def get_query(query_id: str, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]:
|
||||
verified_identity(request, x_tenant_id)
|
||||
raise AnalyticsError("NOT_FOUND", f"query {query_id} is not persisted", 404)
|
||||
|
||||
|
||||
@app.post("/api/0pt/analytics/query/{query_id}/cancel")
|
||||
def cancel_query(query_id: str, request: Request, x_tenant_id: str | None = Header(default=None)) -> dict[str, Any]:
|
||||
verified_identity(request, x_tenant_id)
|
||||
raise AnalyticsError("NOT_FOUND", f"query {query_id} is not running", 404)
|
||||
@@ -0,0 +1,154 @@
|
||||
"""
|
||||
Backfill/refresh: AUD0 events -> 0pt ClickHouse analytics events.
|
||||
|
||||
One audit event becomes one analytics event with metric `event_count` and
|
||||
dimensions `event_type`, `app_id`, `capability_id`, `risk_level`. The write is
|
||||
idempotent per event_id: already-present ids are skipped before the batch is
|
||||
inserted. No synthetic event is ever created; an empty AUD0 page writes
|
||||
nothing.
|
||||
|
||||
Usage:
|
||||
AUD0_TOKEN=... AUD0_TENANT_ID=379... CLICKHOUSE_URL=http://... \
|
||||
python ingest_aud0.py --limit 50000
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from typing import Any, Iterable
|
||||
|
||||
import httpx
|
||||
|
||||
AUD0_API_URL = os.getenv("AUD0_API_URL", "https://api.aud0.siax.io").rstrip("/")
|
||||
AUD0_TOKEN = os.getenv("AUD0_TOKEN", "")
|
||||
AUD0_TENANT_ID = os.getenv("AUD0_TENANT_ID", "")
|
||||
CLICKHOUSE_URL = os.getenv("CLICKHOUSE_URL", "").rstrip("/")
|
||||
CLICKHOUSE_USER = os.getenv("CLICKHOUSE_USER", "default")
|
||||
CLICKHOUSE_PASSWORD = os.getenv("CLICKHOUSE_PASSWORD", "")
|
||||
CLICKHOUSE_DATABASE = os.getenv("CLICKHOUSE_DATABASE", "opt_analytics")
|
||||
|
||||
|
||||
def dims(event: dict[str, Any]) -> tuple[list[str], list[str]]:
|
||||
names: list[str] = []
|
||||
values: list[str] = []
|
||||
for key in ("event_type", "app_id", "capability_id", "risk_level"):
|
||||
value = event.get(key)
|
||||
if isinstance(value, str) and value:
|
||||
names.append(key)
|
||||
values.append(value)
|
||||
return names, values
|
||||
|
||||
|
||||
def to_row(tenant: str, event: dict[str, Any]) -> dict[str, Any] | None:
|
||||
event_id = event.get("event_id")
|
||||
created_at = event.get("created_at")
|
||||
event_type = event.get("event_type")
|
||||
if not isinstance(event_id, str) or not event_id:
|
||||
return None
|
||||
if not isinstance(created_at, str) or not created_at:
|
||||
return None
|
||||
if not isinstance(event_type, str) or not event_type:
|
||||
return None
|
||||
names, values = dims(event)
|
||||
return {
|
||||
"tenant_id": tenant,
|
||||
"project_id": "aud0",
|
||||
"dataset_id": "aud0.events",
|
||||
"metric": "event_count",
|
||||
"event_id": event_id,
|
||||
"event_time": created_at,
|
||||
"value": 1,
|
||||
"dim_names": names,
|
||||
"dim_values": values,
|
||||
}
|
||||
|
||||
|
||||
def ch_insert(rows: Iterable[dict[str, Any]]) -> None:
|
||||
payload = "\n".join(json.dumps(row, ensure_ascii=False) for row in rows)
|
||||
if not payload:
|
||||
return
|
||||
auth = (CLICKHOUSE_USER, CLICKHOUSE_PASSWORD) if CLICKHOUSE_PASSWORD else None
|
||||
with httpx.Client(base_url=CLICKHOUSE_URL, auth=auth, timeout=60.0) as client:
|
||||
response = client.post(
|
||||
"/",
|
||||
params={"database": CLICKHOUSE_DATABASE, "query": "INSERT INTO opt_analytics_event FORMAT JSONEachRow"},
|
||||
content=payload,
|
||||
)
|
||||
response.raise_for_status()
|
||||
|
||||
|
||||
def existing_ids(tenant: str, ids: list[str]) -> set[str]:
|
||||
if not ids:
|
||||
return set()
|
||||
auth = (CLICKHOUSE_USER, CLICKHOUSE_PASSWORD) if CLICKHOUSE_PASSWORD else None
|
||||
found: set[str] = set()
|
||||
sql = (
|
||||
f"SELECT event_id FROM {CLICKHOUSE_DATABASE}.opt_analytics_event "
|
||||
"WHERE tenant_id = {tenant:String} AND event_id IN ({ids:Array(String)}) FORMAT JSON"
|
||||
)
|
||||
with httpx.Client(base_url=CLICKHOUSE_URL, auth=auth, timeout=30.0) as client:
|
||||
# ClickHouse query parameters travel in the URL; 1000 ids overflow it.
|
||||
for start in range(0, len(ids), 50):
|
||||
chunk = ids[start : start + 50]
|
||||
response = client.post(
|
||||
"/",
|
||||
params={
|
||||
"database": CLICKHOUSE_DATABASE,
|
||||
"param_tenant": tenant,
|
||||
"param_ids": "[" + ",".join("'" + value.replace("'", "\\'") + "'" for value in chunk) + "]",
|
||||
},
|
||||
content=sql,
|
||||
)
|
||||
response.raise_for_status()
|
||||
found.update(row["event_id"] for row in response.json().get("data", []))
|
||||
return found
|
||||
|
||||
|
||||
def fetch_page(offset: int, limit: int) -> dict[str, Any]:
|
||||
if not AUD0_TOKEN or not AUD0_TENANT_ID:
|
||||
raise SystemExit("AUD0_TOKEN and AUD0_TENANT_ID are required")
|
||||
with httpx.Client(base_url=AUD0_API_URL, timeout=30.0) as client:
|
||||
response = client.get(
|
||||
"/v1/aud0/events",
|
||||
params={"tenant_id": AUD0_TENANT_ID, "limit": limit, "offset": offset},
|
||||
headers={"authorization": f"Bearer {AUD0_TOKEN}", "x-tenant-id": AUD0_TENANT_ID},
|
||||
)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--limit", type=int, default=50_000, help="maximum events to read")
|
||||
parser.add_argument("--page", type=int, default=1000)
|
||||
parser.add_argument("--dry-run", action="store_true")
|
||||
args = parser.parse_args()
|
||||
|
||||
if not CLICKHOUSE_URL:
|
||||
raise SystemExit("CLICKHOUSE_URL is required")
|
||||
|
||||
written = skipped = seen = 0
|
||||
for offset in range(0, args.limit, args.page):
|
||||
page = fetch_page(offset, min(args.page, args.limit - offset))
|
||||
events = page.get("events") or []
|
||||
if not events:
|
||||
break
|
||||
rows = [row for row in (to_row(AUD0_TENANT_ID, event) for event in events) if row]
|
||||
present = existing_ids(AUD0_TENANT_ID, [row["event_id"] for row in rows])
|
||||
fresh = [row for row in rows if row["event_id"] not in present]
|
||||
skipped += len(rows) - len(fresh)
|
||||
seen += len(events)
|
||||
if not args.dry_run:
|
||||
ch_insert(fresh)
|
||||
written += len(fresh)
|
||||
print(json.dumps({"offset": offset, "read": len(events), "written": len(fresh), "skipped": skipped}))
|
||||
if len(events) < args.page:
|
||||
break
|
||||
print(json.dumps({"seen": seen, "written": written, "skipped": skipped, "dry_run": args.dry_run}))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,5 @@
|
||||
fastapi==0.115.6
|
||||
uvicorn[standard]==0.34.0
|
||||
httpx==0.28.1
|
||||
PyJWT[crypto]==2.14.0
|
||||
pytest==8.3.4
|
||||
@@ -0,0 +1,52 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||||
import app as analytics # noqa: E402
|
||||
|
||||
|
||||
def test_build_query_forces_tenant_and_dimension_whitelist():
|
||||
sql, params = analytics.build_query(
|
||||
tenant="379",
|
||||
dataset="aud0.events",
|
||||
metrics=["event_count"],
|
||||
dimensions=["event_type"],
|
||||
filters={"app_id": "portal"},
|
||||
from_time="2026-01-01T00:00:00Z",
|
||||
to_time=None,
|
||||
limit=50,
|
||||
)
|
||||
assert "tenant_id = {tenant:String}" in sql
|
||||
assert "dataset_id = {dataset:String}" in sql
|
||||
assert "dim_values[indexOf(dim_names, 'event_type')]" in sql
|
||||
assert params["tenant"] == "379"
|
||||
assert params["dataset"] == "aud0.events"
|
||||
assert params["metric0"] == "event_count"
|
||||
assert params["filter0"] == "portal"
|
||||
with pytest.raises(analytics.AnalyticsError):
|
||||
analytics.build_query(
|
||||
tenant="379",
|
||||
dataset=None,
|
||||
metrics=["event_count"],
|
||||
dimensions=["drop table"],
|
||||
filters=None,
|
||||
from_time=None,
|
||||
to_time=None,
|
||||
limit=10,
|
||||
)
|
||||
|
||||
|
||||
def test_data_routes_require_a_verified_identity():
|
||||
client = TestClient(analytics.app)
|
||||
assert client.get("/api/0pt/analytics/datasets").status_code == 401
|
||||
assert client.get("/api/0pt/analytics/metrics").status_code == 401
|
||||
|
||||
|
||||
def test_health_is_always_a_liveness_answer():
|
||||
client = TestClient(analytics.app)
|
||||
response = client.get("/health")
|
||||
assert response.status_code == 200
|
||||
assert "providers" in response.json()
|
||||
File diff suppressed because one or more lines are too long
@@ -1,4 +1,4 @@
|
||||
# 0pt — cosign-signeringskedjan (2026-09-29)
|
||||
|
||||
Prod-imagen git.cloud.siax.io/sax3l/0pt:prod-6bbd333a22 (sha256:7cbc75bb363c1ee47f8aa86cfbaf996e648e4a0c0c5c5f79a09b87fef93db03b) är cosign-signerad
|
||||
Prod-imagen git.cloud.siax.io/sax3l/0pt:prod-8b5e23aa9b (sha256:889753b7e5a9db3e1c870eeb38f450cb0f20498e1025edb28153233e0b06f46d) är cosign-signerad
|
||||
med nyckelparet vars pubkey ligger här. Privata nyckeln raderad efter signering.
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
-----BEGIN PUBLIC KEY-----
|
||||
MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAELOI331jmeMp5t0XySt0ls4XH0Msr
|
||||
2k7neTq+QtRAOuLL4b5Nj3xdHLpDBjaX82zI6XXbJRixh0TeWrp0Y0SF3Q==
|
||||
MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAEiPRG9iG1pjIW548KoIduQPD/B3Yb
|
||||
tg5hPmy7o4feVQsdDnW38VtemVBQ4I+goytV9X4XkrFiweKmfqEMhGyPgg==
|
||||
-----END PUBLIC KEY-----
|
||||
|
||||
@@ -0,0 +1,9 @@
|
||||
{
|
||||
"$schema": "https://git.siax.io/sax3l/siax-standard/raw/branch/main/schema/siax.source-bundle.schema.json",
|
||||
"id": "sb-0pt-13e156ec7e6b",
|
||||
"repo": "0PT",
|
||||
"commitSha": "13e156ec7e6be534de9c28d28ec2daff0355ca7f",
|
||||
"branch": "main",
|
||||
"buildId": "coolify:0pt",
|
||||
"producedAt": "2026-11-04T01:26:48Z"
|
||||
}
|
||||
@@ -0,0 +1,16 @@
|
||||
-- Trial / grant credits: a server-owned, idempotent balance grant that is not
|
||||
-- a payment. Extends the existing kind constraint; safe to re-run.
|
||||
DO $$
|
||||
DECLARE cname text;
|
||||
BEGIN
|
||||
SELECT conname INTO cname
|
||||
FROM pg_constraint
|
||||
WHERE conrelid = 'opt_credit_events'::regclass
|
||||
AND contype = 'c'
|
||||
AND pg_get_constraintdef(oid) LIKE '%kind%';
|
||||
IF cname IS NOT NULL THEN
|
||||
EXECUTE format('ALTER TABLE opt_credit_events DROP CONSTRAINT %I', cname);
|
||||
END IF;
|
||||
ALTER TABLE opt_credit_events ADD CONSTRAINT opt_credit_events_kind_check
|
||||
CHECK (kind IN ('verified_payment', 'grant', 'reserve', 'settle', 'release'));
|
||||
END $$;
|
||||
@@ -0,0 +1,8 @@
|
||||
-- Persist the verified DOC0 receipt on the durable upload attempt so an
|
||||
-- idempotent retry returns the same 201 instead of "outcome unknown".
|
||||
-- Additive and re-runnable.
|
||||
ALTER TABLE opt_document_upload_attempts ADD COLUMN IF NOT EXISTS document_id text;
|
||||
ALTER TABLE opt_document_upload_attempts ADD COLUMN IF NOT EXISTS version_id text;
|
||||
ALTER TABLE opt_document_upload_attempts ADD COLUMN IF NOT EXISTS receipt_checksum text;
|
||||
ALTER TABLE opt_document_upload_attempts ADD COLUMN IF NOT EXISTS pipeline_stage text;
|
||||
ALTER TABLE opt_document_upload_attempts ADD COLUMN IF NOT EXISTS verified_at timestamptz;
|
||||
+25
-8
@@ -50,6 +50,7 @@ import {
|
||||
type JWTPayload,
|
||||
type JWTVerifyGetKey,
|
||||
} from 'jose';
|
||||
import { normalizeZitadelClaims, ZitadelClaimError } from '@siax/auth-zitadel';
|
||||
|
||||
/** Default fleet Zitadel issuer (OIDC). Overridable via ZITADEL_ISSUER. */
|
||||
export const DEFAULT_ZITADEL_ISSUER = 'https://id-customers.siax.io';
|
||||
@@ -254,13 +255,29 @@ export async function verifyBearer(
|
||||
throw new BearerVerificationError('token_verification_failed', err);
|
||||
}
|
||||
|
||||
const sub = typeof payload.sub === 'string' ? payload.sub : undefined;
|
||||
if (!sub || sub.length === 0) {
|
||||
throw new BearerVerificationError('token_missing_sub');
|
||||
try {
|
||||
const identity = normalizeZitadelClaims(payload, {
|
||||
source: 'jwt',
|
||||
tenantClaim: opts.tenantClaim ?? tenantClaimOverride(),
|
||||
});
|
||||
return {
|
||||
sub: identity.principal.subject,
|
||||
tenantId: identity.tenant.tenantId,
|
||||
claims: payload,
|
||||
};
|
||||
} catch (err) {
|
||||
if (err instanceof ZitadelClaimError) {
|
||||
if (/sub/i.test(err.reason)) {
|
||||
throw new BearerVerificationError('token_missing_sub');
|
||||
}
|
||||
if (/tenant/i.test(err.reason)) {
|
||||
throw new BearerVerificationError('token_missing_tenant_claim');
|
||||
}
|
||||
if (/aud/i.test(err.reason)) {
|
||||
throw new BearerVerificationError('token_invalid_audience');
|
||||
}
|
||||
throw new BearerVerificationError('token_verification_failed', err);
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
const tenantId = deriveTenantId(payload, opts.tenantClaim ?? tenantClaimOverride());
|
||||
if (!tenantId) {
|
||||
throw new BearerVerificationError('token_missing_tenant_claim');
|
||||
}
|
||||
return { sub, tenantId, claims: payload };
|
||||
}
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import type { RepositorySql } from '../repositories/db';
|
||||
import { withCreditTenant } from './credit-tenant-scope';
|
||||
|
||||
export type CreditOperation = 'verified_payment' | 'reserve' | 'settle' | 'release';
|
||||
export type CreditOperation = 'verified_payment' | 'grant' | 'reserve' | 'settle' | 'release';
|
||||
export interface CreditCommand {
|
||||
tenantId: string;
|
||||
idempotencyKey: string;
|
||||
@@ -26,8 +26,8 @@ export class PostgresCreditLedger {
|
||||
|
||||
/** Read-only balance for the authenticated tenant. Never creates a row. */
|
||||
async balance(tenantId: string): Promise<CreditBalance> {
|
||||
if (typeof tenantId !== "string" || !tenantId.trim() || tenantId.length > 256) {
|
||||
throw new Error("Invalid credit tenant");
|
||||
if (typeof tenantId !== 'string' || !tenantId.trim() || tenantId.length > 256) {
|
||||
throw new Error('Invalid credit tenant');
|
||||
}
|
||||
return withCreditTenant(this.sql, tenantId, async (tx) => {
|
||||
const rows = await tx<{ available: string; reserved: string }[]>`
|
||||
@@ -45,10 +45,10 @@ export class PostgresCreditLedger {
|
||||
![tenantId, idempotencyKey, reference].every(
|
||||
(value) => typeof value === 'string' && value.trim().length > 0,
|
||||
) ||
|
||||
!['verified_payment', 'reserve', 'settle', 'release'].includes(kind) ||
|
||||
!['verified_payment', 'grant', 'reserve', 'settle', 'release'].includes(kind) ||
|
||||
!Number.isSafeInteger(amount) ||
|
||||
amount < 0 ||
|
||||
((kind === 'verified_payment' || kind === 'reserve') && amount === 0) ||
|
||||
((kind === 'verified_payment' || kind === 'grant' || kind === 'reserve') && amount === 0) ||
|
||||
(kind === 'release' && amount !== 0)
|
||||
)
|
||||
throw new Error('Invalid credit command');
|
||||
@@ -73,7 +73,7 @@ export class PostgresCreditLedger {
|
||||
return { replay: true, balance: { available, reserved } };
|
||||
}
|
||||
|
||||
if (kind === 'verified_payment') {
|
||||
if (kind === 'verified_payment' || kind === 'grant') {
|
||||
available += amount;
|
||||
} else if (kind === 'reserve') {
|
||||
if (available < amount) throw new Error('Insufficient available credits');
|
||||
|
||||
@@ -0,0 +1,40 @@
|
||||
/**
|
||||
* One-time trial credits.
|
||||
*
|
||||
* A new tenant can be granted a fixed, server-owned amount so the credit
|
||||
* ledger (reserve → settle) works in production without a card charge. It is
|
||||
* idempotent per tenant: the ledger's unique (tenant, idempotency_key) makes
|
||||
* every repeat a replay, never a second grant. The amount is never read from a
|
||||
* client; `OPT_TRIAL_CREDITS=0` (default) disables it entirely.
|
||||
*/
|
||||
export interface CreditCommandSink {
|
||||
apply(command: {
|
||||
tenantId: string;
|
||||
idempotencyKey: string;
|
||||
kind: 'grant';
|
||||
reference: string;
|
||||
amount: number;
|
||||
}): Promise<unknown>;
|
||||
}
|
||||
|
||||
export function readTrialCredits(env: NodeJS.ProcessEnv = process.env): number {
|
||||
const parsed = Number(env.OPT_TRIAL_CREDITS ?? '0');
|
||||
if (!Number.isSafeInteger(parsed) || parsed < 0 || parsed > 1_000_000) return 0;
|
||||
return parsed;
|
||||
}
|
||||
|
||||
export async function ensureTrialCredits(
|
||||
ledger: CreditCommandSink,
|
||||
tenantId: string,
|
||||
amount: number,
|
||||
): Promise<void> {
|
||||
if (!Number.isSafeInteger(amount) || amount <= 0) return;
|
||||
if (typeof tenantId !== 'string' || !tenantId.trim()) return;
|
||||
await ledger.apply({
|
||||
tenantId,
|
||||
idempotencyKey: `trial:${tenantId}`,
|
||||
kind: 'grant',
|
||||
reference: `trial:${tenantId}`,
|
||||
amount,
|
||||
});
|
||||
}
|
||||
@@ -13,6 +13,12 @@ export interface UploadAttemptIdentity {
|
||||
requestFingerprint: string;
|
||||
}
|
||||
export type UploadAttemptState = 'prepared' | 'dispatch_started';
|
||||
export interface StoredUploadReceipt {
|
||||
documentId: string;
|
||||
versionId: string;
|
||||
checksumSha256: string;
|
||||
pipelineStage: string;
|
||||
}
|
||||
interface Row {
|
||||
source_sha256: string;
|
||||
source_size_bytes: string;
|
||||
@@ -21,10 +27,15 @@ interface Row {
|
||||
}
|
||||
|
||||
function validate(input: UploadAttemptIdentity): void {
|
||||
if (![input.tenantId, input.projectId, input.actorId, input.attemptId].every(
|
||||
(value) => typeof value === 'string' && value.trim().length > 0 && value.length <= 256,
|
||||
) || !/^[0-9a-f]{64}$/.test(input.sourceSha256) || !/^[0-9a-f]{64}$/.test(input.requestFingerprint)
|
||||
|| !Number.isSafeInteger(input.sourceSizeBytes) || input.sourceSizeBytes <= 0) {
|
||||
if (
|
||||
![input.tenantId, input.projectId, input.actorId, input.attemptId].every(
|
||||
(value) => typeof value === 'string' && value.trim().length > 0 && value.length <= 256,
|
||||
) ||
|
||||
!/^[0-9a-f]{64}$/.test(input.sourceSha256) ||
|
||||
!/^[0-9a-f]{64}$/.test(input.requestFingerprint) ||
|
||||
!Number.isSafeInteger(input.sourceSizeBytes) ||
|
||||
input.sourceSizeBytes <= 0
|
||||
) {
|
||||
throw new Error('Invalid document upload attempt');
|
||||
}
|
||||
}
|
||||
@@ -40,12 +51,16 @@ function validate(input: UploadAttemptIdentity): void {
|
||||
export class PostgresUploadAttemptStore {
|
||||
constructor(private readonly sql: RepositorySql) {}
|
||||
|
||||
private async scoped<T>(input: UploadAttemptIdentity, action: (tx: postgres.TransactionSql) => Promise<T>): Promise<T> {
|
||||
private async scoped<T>(
|
||||
input: UploadAttemptIdentity,
|
||||
action: (tx: postgres.TransactionSql) => Promise<T>,
|
||||
): Promise<T> {
|
||||
validate(input);
|
||||
return this.sql.begin(async (tx) => {
|
||||
const roles = await tx<{ rolsuper: boolean; rolbypassrls: boolean }[]>`
|
||||
SELECT rolsuper, rolbypassrls FROM pg_roles WHERE rolname = current_user`;
|
||||
if (!roles[0] || roles[0].rolsuper || roles[0].rolbypassrls) throw new Error('Upload attempts require an RLS-enforced database role');
|
||||
if (!roles[0] || roles[0].rolsuper || roles[0].rolbypassrls)
|
||||
throw new Error('Upload attempts require an RLS-enforced database role');
|
||||
await tx`SELECT set_config('app.opt_tenant_id', ${input.tenantId}, true), set_config('app.opt_project_id', ${input.projectId}, true), set_config('app.opt_actor_id', ${input.actorId}, true)`;
|
||||
return action(tx);
|
||||
}) as Promise<T>;
|
||||
@@ -57,11 +72,18 @@ export class PostgresUploadAttemptStore {
|
||||
INSERT INTO opt_document_upload_attempts (tenant_id,project_id,actor_id,attempt_id,source_sha256,source_size_bytes,request_fingerprint,state)
|
||||
VALUES (${input.tenantId},${input.projectId},${input.actorId},${input.attemptId},${input.sourceSha256},${input.sourceSizeBytes},${input.requestFingerprint},'prepared')
|
||||
ON CONFLICT (tenant_id,attempt_id) DO NOTHING RETURNING source_sha256,source_size_bytes,request_fingerprint,state`;
|
||||
const rows = inserted.length ? inserted : await tx<Row[]>`
|
||||
const rows = inserted.length
|
||||
? inserted
|
||||
: await tx<Row[]>`
|
||||
SELECT source_sha256,source_size_bytes,request_fingerprint,state FROM opt_document_upload_attempts
|
||||
WHERE tenant_id=${input.tenantId} AND project_id=${input.projectId} AND actor_id=${input.actorId} AND attempt_id=${input.attemptId}`;
|
||||
const row = rows[0];
|
||||
if (!row || row.source_sha256 !== input.sourceSha256 || row.source_size_bytes !== String(input.sourceSizeBytes) || row.request_fingerprint !== input.requestFingerprint) {
|
||||
if (
|
||||
!row ||
|
||||
row.source_sha256 !== input.sourceSha256 ||
|
||||
row.source_size_bytes !== String(input.sourceSizeBytes) ||
|
||||
row.request_fingerprint !== input.requestFingerprint
|
||||
) {
|
||||
throw new Error('Document upload attempt cannot be reused');
|
||||
}
|
||||
return { state: row.state, replay: inserted.length === 0 };
|
||||
@@ -80,4 +102,59 @@ export class PostgresUploadAttemptStore {
|
||||
return rows.length === 1;
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Persist the verified DOC0 receipt so a retry with the same idempotency key
|
||||
* returns the SAME 201 receipt instead of "outcome unknown". Only the fields
|
||||
* DOC0 itself returned are stored; storage/extraction are not re-proven here.
|
||||
*/
|
||||
async recordReceipt(input: UploadAttemptIdentity, receipt: StoredUploadReceipt): Promise<void> {
|
||||
if (
|
||||
![receipt.documentId, receipt.versionId, receipt.pipelineStage].every(
|
||||
(value) => typeof value === 'string' && value.trim().length > 0 && value.length <= 256,
|
||||
) ||
|
||||
!/^[0-9a-f]{64}$/.test(receipt.checksumSha256)
|
||||
) {
|
||||
throw new Error('Invalid upload receipt');
|
||||
}
|
||||
await this.scoped(input, async (tx) => {
|
||||
const rows = await tx<{ attempt_id: string }[]>`
|
||||
UPDATE opt_document_upload_attempts
|
||||
SET document_id=${receipt.documentId}, version_id=${receipt.versionId},
|
||||
receipt_checksum=${receipt.checksumSha256}, pipeline_stage=${receipt.pipelineStage},
|
||||
verified_at=clock_timestamp(), updated_at=clock_timestamp()
|
||||
WHERE tenant_id=${input.tenantId} AND project_id=${input.projectId} AND actor_id=${input.actorId}
|
||||
AND attempt_id=${input.attemptId} AND source_sha256=${input.sourceSha256}
|
||||
AND source_size_bytes=${input.sourceSizeBytes} AND request_fingerprint=${input.requestFingerprint}
|
||||
RETURNING attempt_id`;
|
||||
if (rows.length !== 1) throw new Error('Upload attempt receipt could not be recorded');
|
||||
});
|
||||
}
|
||||
|
||||
async getReceipt(input: UploadAttemptIdentity): Promise<StoredUploadReceipt | undefined> {
|
||||
return this.scoped(input, async (tx) => {
|
||||
const rows = await tx<
|
||||
{
|
||||
document_id: string | null;
|
||||
version_id: string | null;
|
||||
receipt_checksum: string | null;
|
||||
pipeline_stage: string | null;
|
||||
}[]
|
||||
>`
|
||||
SELECT document_id, version_id, receipt_checksum, pipeline_stage
|
||||
FROM opt_document_upload_attempts
|
||||
WHERE tenant_id=${input.tenantId} AND project_id=${input.projectId} AND actor_id=${input.actorId}
|
||||
AND attempt_id=${input.attemptId} AND source_sha256=${input.sourceSha256}
|
||||
AND source_size_bytes=${input.sourceSizeBytes} AND request_fingerprint=${input.requestFingerprint}`;
|
||||
const row = rows[0];
|
||||
if (!row?.document_id || !row.version_id || !row.receipt_checksum || !row.pipeline_stage)
|
||||
return undefined;
|
||||
return {
|
||||
documentId: row.document_id,
|
||||
versionId: row.version_id,
|
||||
checksumSha256: row.receipt_checksum,
|
||||
pipelineStage: row.pipeline_stage,
|
||||
};
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -6,12 +6,15 @@ import {
|
||||
type Doc0UploadInput,
|
||||
type Doc0UploadReceipt,
|
||||
} from './doc0-upload-client';
|
||||
import type { UploadAttemptIdentity, UploadAttemptState } from './upload-attempt-store';
|
||||
import type { StoredUploadReceipt, UploadAttemptIdentity, UploadAttemptState } from './upload-attempt-store';
|
||||
|
||||
type Scope = Pick<Doc0UploadInput, 'tenantId' | 'projectId' | 'actorId'>;
|
||||
type Attempts = {
|
||||
prepare(input: UploadAttemptIdentity): Promise<{ state: UploadAttemptState; replay: boolean }>;
|
||||
startDispatch(input: UploadAttemptIdentity): Promise<boolean>;
|
||||
/** Optional so lightweight tests/implementations keep working. */
|
||||
recordReceipt?(input: UploadAttemptIdentity, receipt: StoredUploadReceipt): Promise<void>;
|
||||
getReceipt?(input: UploadAttemptIdentity): Promise<StoredUploadReceipt | undefined>;
|
||||
};
|
||||
type Doc0 = {
|
||||
upload(input: Doc0UploadInput): Promise<Doc0UploadReceipt>;
|
||||
@@ -28,12 +31,14 @@ export type CoordinatedUpload =
|
||||
* be deployed and verified before this coordinator can serve customer data.
|
||||
*/
|
||||
export class DocumentUploadCoordinator {
|
||||
constructor(private readonly deps: {
|
||||
attempts: Attempts;
|
||||
doc0: Doc0;
|
||||
authorize: (scope: Scope) => Promise<boolean>;
|
||||
maxBytes: number;
|
||||
}) {
|
||||
constructor(
|
||||
private readonly deps: {
|
||||
attempts: Attempts;
|
||||
doc0: Doc0;
|
||||
authorize: (scope: Scope) => Promise<boolean>;
|
||||
maxBytes: number;
|
||||
},
|
||||
) {
|
||||
if (!Number.isSafeInteger(deps.maxBytes) || deps.maxBytes <= 0) throw new Error('Invalid upload limit');
|
||||
}
|
||||
|
||||
@@ -48,27 +53,50 @@ export class DocumentUploadCoordinator {
|
||||
const scope = { tenantId: snapshot.tenantId, projectId: snapshot.projectId, actorId: snapshot.actorId };
|
||||
if ((await this.deps.authorize(scope)) !== true) throw new Error('Document upload not authorized');
|
||||
const attempt: UploadAttemptIdentity = {
|
||||
...scope, attemptId: snapshot.idempotencyKey, sourceSha256: details.sourceSha256,
|
||||
sourceSizeBytes: details.sourceSizeBytes, requestFingerprint: details.requestFingerprint,
|
||||
...scope,
|
||||
attemptId: snapshot.idempotencyKey,
|
||||
sourceSha256: details.sourceSha256,
|
||||
sourceSizeBytes: details.sourceSizeBytes,
|
||||
requestFingerprint: details.requestFingerprint,
|
||||
};
|
||||
await this.deps.attempts.prepare(attempt);
|
||||
// A database error here also stops the operation: the update might have
|
||||
// committed and must never be followed by an unrecorded network write.
|
||||
if ((await this.deps.attempts.startDispatch(attempt)) === true) {
|
||||
try {
|
||||
return { status: 'receipt_verified', receipt: await this.deps.doc0.upload(snapshot) };
|
||||
const receipt = await this.deps.doc0.upload(snapshot);
|
||||
// Persist the verified receipt so retrying with the same idempotency
|
||||
// key returns this exact 201 instead of "outcome unknown". A journal
|
||||
// write failure must not turn a real upload into an error response.
|
||||
try {
|
||||
await this.deps.attempts.recordReceipt?.(attempt, receipt);
|
||||
} catch {
|
||||
/* receipt stays unpersisted; a retry falls back to observation */
|
||||
}
|
||||
return { status: 'receipt_verified', receipt };
|
||||
} catch (error) {
|
||||
// The client marks every potentially committed POST as unknown. Any
|
||||
// other error propagates, while the one-way dispatch remains recorded.
|
||||
if (!(error instanceof Doc0UploadOutcomeUnknownError)) throw error;
|
||||
}
|
||||
}
|
||||
// A retry of an already-dispatched attempt: if the first dispatch returned
|
||||
// a verified receipt, return it again (same document, same checksum).
|
||||
try {
|
||||
const stored = await this.deps.attempts.getReceipt?.(attempt);
|
||||
if (stored) return { status: 'receipt_verified', receipt: stored };
|
||||
} catch {
|
||||
/* fall through to observation */
|
||||
}
|
||||
// Observation only. Even not_found does not prove that a write cannot
|
||||
// still arrive; completed_recorded does not prove storage or extraction.
|
||||
try {
|
||||
const observation = await this.deps.doc0.inspect({
|
||||
...scope, requestId: snapshot.requestId, idempotencyKey: snapshot.idempotencyKey,
|
||||
sourceSha256: details.sourceSha256, sourceSizeBytes: details.sourceSizeBytes,
|
||||
...scope,
|
||||
requestId: snapshot.requestId,
|
||||
idempotencyKey: snapshot.idempotencyKey,
|
||||
sourceSha256: details.sourceSha256,
|
||||
sourceSizeBytes: details.sourceSizeBytes,
|
||||
requestFingerprint: details.requestFingerprint,
|
||||
});
|
||||
return { status: 'outcome_unknown', observation };
|
||||
|
||||
@@ -40,6 +40,7 @@ async function sign(claims: Record<string, unknown>, opts: SignOpts = {}): Promi
|
||||
.setProtectedHeader({ alg: 'RS256', kid: KID })
|
||||
.setIssuedAt()
|
||||
.setIssuer(opts.issuer ?? ISSUER)
|
||||
.setAudience('siax-test')
|
||||
.setExpirationTime(opts.expirationTime ?? '2h')
|
||||
.sign(privateKey);
|
||||
}
|
||||
|
||||
@@ -0,0 +1,25 @@
|
||||
import { describe, expect, it, vi } from 'vitest';
|
||||
import { ensureTrialCredits, readTrialCredits } from '../../src/document-workbench/trial-credits';
|
||||
|
||||
describe('trial credits', () => {
|
||||
it('is disabled by default and rejects invalid amounts', () => {
|
||||
expect(readTrialCredits({} as unknown as NodeJS.ProcessEnv)).toBe(0);
|
||||
expect(readTrialCredits({ OPT_TRIAL_CREDITS: '-5' } as unknown as NodeJS.ProcessEnv)).toBe(0);
|
||||
expect(readTrialCredits({ OPT_TRIAL_CREDITS: 'abc' } as unknown as NodeJS.ProcessEnv)).toBe(0);
|
||||
expect(readTrialCredits({ OPT_TRIAL_CREDITS: '100' } as unknown as NodeJS.ProcessEnv)).toBe(100);
|
||||
});
|
||||
|
||||
it('applies one idempotent grant and does nothing for zero', async () => {
|
||||
const apply = vi.fn(async () => ({ replay: false }));
|
||||
await ensureTrialCredits({ apply }, 't_0pt', 100);
|
||||
expect(apply).toHaveBeenCalledWith({
|
||||
tenantId: 't_0pt',
|
||||
idempotencyKey: 'trial:t_0pt',
|
||||
kind: 'grant',
|
||||
reference: 'trial:t_0pt',
|
||||
amount: 100,
|
||||
});
|
||||
await ensureTrialCredits({ apply }, 't_0pt', 0);
|
||||
expect(apply).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
||||
@@ -1,15 +1,38 @@
|
||||
import { describe, expect, it, vi } from 'vitest';
|
||||
import { Doc0UploadOutcomeUnknownError, prepareDoc0UploadIdentity, type Doc0UploadClient } from '../../src/document-workbench/doc0-upload-client';
|
||||
import {
|
||||
Doc0UploadOutcomeUnknownError,
|
||||
prepareDoc0UploadIdentity,
|
||||
type Doc0UploadClient,
|
||||
} from '../../src/document-workbench/doc0-upload-client';
|
||||
import { DocumentUploadCoordinator } from '../../src/document-workbench/upload-coordinator';
|
||||
import type { PostgresUploadAttemptStore } from '../../src/document-workbench/upload-attempt-store';
|
||||
|
||||
const input = { tenantId: 't1', projectId: 'p1', actorId: 'a1', requestId: 'r1', idempotencyKey: 'i1',
|
||||
filename: 'rapport åäö.txt', mimeType: 'text/plain', classification: 'confidential' as const,
|
||||
bytes: new TextEncoder().encode('original') };
|
||||
const input = {
|
||||
tenantId: 't1',
|
||||
projectId: 'p1',
|
||||
actorId: 'a1',
|
||||
requestId: 'r1',
|
||||
idempotencyKey: 'i1',
|
||||
filename: 'rapport åäö.txt',
|
||||
mimeType: 'text/plain',
|
||||
classification: 'confidential' as const,
|
||||
bytes: new TextEncoder().encode('original'),
|
||||
};
|
||||
const receipt = { documentId: 'd1', versionId: 'v1', checksumSha256: 'test', pipelineStage: 'accepted' };
|
||||
function fixture() {
|
||||
const attempts = { prepare: vi.fn<PostgresUploadAttemptStore['prepare']>(async () => ({ state: 'prepared' as const, replay: false })), startDispatch: vi.fn<PostgresUploadAttemptStore['startDispatch']>(async () => true) };
|
||||
const doc0 = { upload: vi.fn<Doc0UploadClient['upload']>(async () => receipt), inspect: vi.fn<Doc0UploadClient['inspect']>(async () => ({ status: 'not_found' as const })) };
|
||||
const attempts = {
|
||||
prepare: vi.fn<PostgresUploadAttemptStore['prepare']>(async () => ({
|
||||
state: 'prepared' as const,
|
||||
replay: false,
|
||||
})),
|
||||
startDispatch: vi.fn<PostgresUploadAttemptStore['startDispatch']>(async () => true),
|
||||
recordReceipt: vi.fn<PostgresUploadAttemptStore['recordReceipt']>(async () => undefined),
|
||||
getReceipt: vi.fn<PostgresUploadAttemptStore['getReceipt']>(async () => undefined),
|
||||
};
|
||||
const doc0 = {
|
||||
upload: vi.fn<Doc0UploadClient['upload']>(async () => receipt),
|
||||
inspect: vi.fn<Doc0UploadClient['inspect']>(async () => ({ status: 'not_found' as const })),
|
||||
};
|
||||
const authorize = vi.fn(async () => true);
|
||||
const coordinator = new DocumentUploadCoordinator({ attempts, doc0, authorize, maxBytes: 100 });
|
||||
return { attempts, doc0, authorize, coordinator };
|
||||
@@ -20,47 +43,99 @@ describe('document upload coordination', () => {
|
||||
expect(await f.coordinator.execute(input)).toEqual({ status: 'receipt_verified', receipt });
|
||||
expect(f.authorize).toHaveBeenCalledWith({ tenantId: 't1', projectId: 'p1', actorId: 'a1' });
|
||||
const attempt = f.attempts.prepare.mock.calls[0][0];
|
||||
expect(attempt).toEqual({ tenantId: 't1', projectId: 'p1', actorId: 'a1', attemptId: 'i1',
|
||||
sourceSha256: prepareDoc0UploadIdentity(input).sourceSha256, sourceSizeBytes: input.bytes.length,
|
||||
requestFingerprint: prepareDoc0UploadIdentity(input).requestFingerprint });
|
||||
expect(attempt).toEqual({
|
||||
tenantId: 't1',
|
||||
projectId: 'p1',
|
||||
actorId: 'a1',
|
||||
attemptId: 'i1',
|
||||
sourceSha256: prepareDoc0UploadIdentity(input).sourceSha256,
|
||||
sourceSizeBytes: input.bytes.length,
|
||||
requestFingerprint: prepareDoc0UploadIdentity(input).requestFingerprint,
|
||||
});
|
||||
expect(f.attempts.startDispatch).toHaveBeenCalledWith(attempt);
|
||||
expect(f.attempts.prepare.mock.invocationCallOrder[0]).toBeLessThan(f.attempts.startDispatch.mock.invocationCallOrder[0]);
|
||||
expect(f.attempts.startDispatch.mock.invocationCallOrder[0]).toBeLessThan(f.doc0.upload.mock.invocationCallOrder[0]);
|
||||
expect(f.attempts.prepare.mock.invocationCallOrder[0]).toBeLessThan(
|
||||
f.attempts.startDispatch.mock.invocationCallOrder[0],
|
||||
);
|
||||
expect(f.attempts.startDispatch.mock.invocationCallOrder[0]).toBeLessThan(
|
||||
f.doc0.upload.mock.invocationCallOrder[0],
|
||||
);
|
||||
});
|
||||
it('denies before persistence or network when authorization fails', async () => {
|
||||
const f = fixture(); f.authorize.mockResolvedValueOnce(false);
|
||||
const f = fixture();
|
||||
f.authorize.mockResolvedValueOnce(false);
|
||||
await expect(f.coordinator.execute(input)).rejects.toThrow(/not authorized/);
|
||||
expect(f.attempts.prepare).not.toHaveBeenCalled(); expect(f.doc0.upload).not.toHaveBeenCalled();
|
||||
expect(f.attempts.prepare).not.toHaveBeenCalled();
|
||||
expect(f.doc0.upload).not.toHaveBeenCalled();
|
||||
});
|
||||
it('rejects truthy but non-boolean authorization results', async () => {
|
||||
const f = fixture(); f.authorize.mockResolvedValueOnce('allow' as unknown as boolean);
|
||||
const f = fixture();
|
||||
f.authorize.mockResolvedValueOnce('allow' as unknown as boolean);
|
||||
await expect(f.coordinator.execute(input)).rejects.toThrow(/not authorized/);
|
||||
expect(f.attempts.prepare).not.toHaveBeenCalled();
|
||||
});
|
||||
it('does not dispatch after a failed or ambiguous database transition', async () => {
|
||||
const f = fixture(); f.attempts.startDispatch.mockRejectedValueOnce(new Error('connection lost'));
|
||||
const f = fixture();
|
||||
f.attempts.startDispatch.mockRejectedValueOnce(new Error('connection lost'));
|
||||
await expect(f.coordinator.execute(input)).rejects.toThrow(/connection lost/);
|
||||
expect(f.doc0.upload).not.toHaveBeenCalled(); expect(f.doc0.inspect).not.toHaveBeenCalled();
|
||||
expect(f.doc0.upload).not.toHaveBeenCalled();
|
||||
expect(f.doc0.inspect).not.toHaveBeenCalled();
|
||||
});
|
||||
it('only observes a previously started attempt and never sends a second POST', async () => {
|
||||
const f = fixture(); f.attempts.startDispatch.mockResolvedValueOnce(false);
|
||||
expect(await f.coordinator.execute(input)).toEqual({ status: 'outcome_unknown', observation: { status: 'not_found' } });
|
||||
expect(f.doc0.upload).not.toHaveBeenCalled(); expect(f.doc0.inspect).toHaveBeenCalledOnce();
|
||||
const f = fixture();
|
||||
f.attempts.startDispatch.mockResolvedValueOnce(false);
|
||||
expect(await f.coordinator.execute(input)).toEqual({
|
||||
status: 'outcome_unknown',
|
||||
observation: { status: 'not_found' },
|
||||
});
|
||||
expect(f.doc0.upload).not.toHaveBeenCalled();
|
||||
expect(f.doc0.inspect).toHaveBeenCalledOnce();
|
||||
});
|
||||
it('persists the verified receipt after a successful upload', async () => {
|
||||
const f = fixture();
|
||||
expect(await f.coordinator.execute(input)).toEqual({ status: 'receipt_verified', receipt });
|
||||
expect(f.attempts.recordReceipt).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ attemptId: 'i1' }),
|
||||
receipt,
|
||||
);
|
||||
});
|
||||
it('returns the SAME verified receipt on an idempotent retry without a second POST', async () => {
|
||||
const f = fixture();
|
||||
f.attempts.startDispatch.mockResolvedValueOnce(false);
|
||||
f.attempts.getReceipt.mockResolvedValueOnce(receipt);
|
||||
expect(await f.coordinator.execute(input)).toEqual({ status: 'receipt_verified', receipt });
|
||||
expect(f.doc0.upload).not.toHaveBeenCalled();
|
||||
expect(f.doc0.inspect).not.toHaveBeenCalled();
|
||||
});
|
||||
it('keeps a journaled completion distinct from a verified receipt', async () => {
|
||||
const f = fixture(); f.attempts.startDispatch.mockResolvedValueOnce(false);
|
||||
f.doc0.inspect.mockResolvedValueOnce({ status: 'completed_recorded', documentId: 'doc', versionId: 'version' });
|
||||
expect(await f.coordinator.execute(input)).toEqual({ status: 'outcome_unknown', observation: { status: 'completed_recorded', documentId: 'doc', versionId: 'version' } });
|
||||
const f = fixture();
|
||||
f.attempts.startDispatch.mockResolvedValueOnce(false);
|
||||
f.doc0.inspect.mockResolvedValueOnce({
|
||||
status: 'completed_recorded',
|
||||
documentId: 'doc',
|
||||
versionId: 'version',
|
||||
});
|
||||
expect(await f.coordinator.execute(input)).toEqual({
|
||||
status: 'outcome_unknown',
|
||||
observation: { status: 'completed_recorded', documentId: 'doc', versionId: 'version' },
|
||||
});
|
||||
expect(f.doc0.upload).not.toHaveBeenCalled();
|
||||
});
|
||||
it('keeps a possibly committed upload unknown even when inspection is unavailable', async () => {
|
||||
const f = fixture(); f.doc0.upload.mockRejectedValueOnce(new Doc0UploadOutcomeUnknownError()); f.doc0.inspect.mockRejectedValueOnce(new Error('unavailable'));
|
||||
const f = fixture();
|
||||
f.doc0.upload.mockRejectedValueOnce(new Doc0UploadOutcomeUnknownError());
|
||||
f.doc0.inspect.mockRejectedValueOnce(new Error('unavailable'));
|
||||
expect(await f.coordinator.execute(input)).toEqual({ status: 'outcome_unknown' });
|
||||
expect(f.doc0.upload).toHaveBeenCalledOnce(); expect(f.doc0.inspect).toHaveBeenCalledOnce();
|
||||
expect(f.doc0.upload).toHaveBeenCalledOnce();
|
||||
expect(f.doc0.inspect).toHaveBeenCalledOnce();
|
||||
});
|
||||
it('snapshots bytes and metadata before asynchronous authorization', async () => {
|
||||
const f = fixture(); const mutable = { ...input, bytes: new Uint8Array(input.bytes) };
|
||||
f.authorize.mockImplementationOnce(async () => { mutable.bytes.fill(0); mutable.filename = 'changed.txt'; return true; });
|
||||
const f = fixture();
|
||||
const mutable = { ...input, bytes: new Uint8Array(input.bytes) };
|
||||
f.authorize.mockImplementationOnce(async () => {
|
||||
mutable.bytes.fill(0);
|
||||
mutable.filename = 'changed.txt';
|
||||
return true;
|
||||
});
|
||||
await f.coordinator.execute(mutable);
|
||||
expect(f.doc0.upload.mock.calls[0][0].filename).toBe(input.filename);
|
||||
expect(new TextDecoder().decode(f.doc0.upload.mock.calls[0][0].bytes)).toBe('original');
|
||||
|
||||
Reference in New Issue
Block a user