feat(server): W2-B sources, identity resolution, metadata and site workbench
- Sites / accounts / feeds: registry view (/site-definitions), site CRUD with
enable/disable (history stays queryable, marked site_disabled), accounts and
feed URLs stored as secrets (only secret_ref / redacted urlDisplay exposed),
feed enable/disable/reset/fetch; one shared rate limiter + adapter per site.
- Tasks: ingestFeed (FeedFetchState/FeedCursor on feed_source, observation
feed+guid upsert with digest-change updates, source_release site+id
aggregation across feeds, auth failure -> account invalid + decision, no
retry, gap -> feed.gap_suspected -> budgeted backfill snapshot),
resolveObservation (parseRelease -> local recall via title_alias pg_trgm +
accepted identities -> TMDB -> resolveIdentity; auto binds entity /
provisional units / identities / aliases, otherwise identity_candidate pool
with evidence version and bounded review backoff; manual lock respected),
refreshMetadata (metadata_field + field_override, user locks win, paths
untouched), snapshotSite (site_snapshot coverage records).
- Events: feed.due, feed.gap_suspected, release.observed (self-consumed),
release.resolved, decision.created/updated; scheduler in worker start().
- Identity API: entity detail with redirects, manual mapping correction with
identity_revision, merge with identity_redirect, pool list/assign/dismiss/
retry, content unit numbering lock, aliases, external identities.
- Decisions: list/summary (by entity and kind)/detail/resolve; identity
options executed directly.
- Workbench: /sites/{id}/releases (snapshot first, coalesced live search with
backoff, origin + freshness) and /sites/{id}/releases/{rid} (volatile field
recheck times, refresh keeps data on failure).
- Capabilities: catalog.lookup, release.search, release.inspect,
identity.resolve.
- Migrations 0018-0020: site.adapter api, source_release.resolution_locked,
site_snapshot, identity_candidate, resolution_attempt, identity_redirect,
metadata_field, field_override.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
1 parent
829ca4f7ac
commit
a24bda952b
39 files changed
+6861
-15
No files matched your search
@@ -0,0 +1,92 @@
|
||||
import { defineTask, type OutboxRouteFn, type TaskRegistry } from "@umt/tasks";
|
||||
import { z } from "zod";
|
||||
import type { SourcesRuntime } from "./runtime.ts";
|
||||
import { ingestFeed } from "./service/feeds.ts";
|
||||
import { refreshMetadata } from "./service/metadata.ts";
|
||||
import { resolveRelease } from "./service/resolve.ts";
|
||||
import { snapshotSite } from "./service/snapshot.ts";
|
||||
|
||||
// 持久任务:输入只放业务 ID,处理器读业务表。外部请求失败的退避/熔断由适配器状态与业务表承担,
|
||||
// 凭据失效返回正常结果(不重试);只有意外错误(DB、TMDB 网络)抛出交给引擎重试。
|
||||
|
||||
const NETWORK_RETRY = {
|
||||
maximumAttempts: 5,
|
||||
initialInterval: "1s",
|
||||
backoffCoefficient: 2,
|
||||
maximumInterval: "60s",
|
||||
} as const;
|
||||
|
||||
const id = z.string().uuid();
|
||||
|
||||
/** Feed 疑似缺口时的补漏页数预算 */
|
||||
export const BACKFILL_MAX_PAGES = 3;
|
||||
|
||||
export function registerSourcesTasks(registry: TaskRegistry, rt: () => SourcesRuntime): void {
|
||||
registry.register(
|
||||
defineTask({
|
||||
name: "ingestFeed",
|
||||
version: 1,
|
||||
inputSchema: z.object({ feedSourceId: id, manual: z.boolean().optional() }),
|
||||
retryPolicy: NETWORK_RETRY,
|
||||
handler: (t) => t.step.run("ingest", () => ingestFeed(rt(), t.input.feedSourceId, { manual: !!t.input.manual })),
|
||||
}),
|
||||
defineTask({
|
||||
name: "resolveObservation",
|
||||
version: 1,
|
||||
inputSchema: z
|
||||
.object({ observationId: id.nullish(), sourceReleaseId: id.nullish(), force: z.boolean().optional() })
|
||||
.refine((v) => !!v.observationId || !!v.sourceReleaseId, { message: "需要 observationId 或 sourceReleaseId" }),
|
||||
retryPolicy: NETWORK_RETRY,
|
||||
handler: (t) =>
|
||||
t.step.run("resolve", () =>
|
||||
resolveRelease(rt(), {
|
||||
observationId: t.input.observationId ?? undefined,
|
||||
sourceReleaseId: t.input.sourceReleaseId ?? undefined,
|
||||
force: t.input.force,
|
||||
}),
|
||||
),
|
||||
}),
|
||||
defineTask({
|
||||
name: "refreshMetadata",
|
||||
version: 1,
|
||||
inputSchema: z.object({ mediaEntityId: id }),
|
||||
retryPolicy: NETWORK_RETRY,
|
||||
handler: (t) => t.step.run("refresh", () => refreshMetadata(rt(), t.input.mediaEntityId)),
|
||||
}),
|
||||
defineTask({
|
||||
name: "snapshotSite",
|
||||
version: 1,
|
||||
inputSchema: z.object({
|
||||
siteId: id,
|
||||
kind: z.enum(["periodic", "backfill", "manual"]).default("periodic"),
|
||||
maxPages: z.number().int().min(1).max(20).optional(),
|
||||
}),
|
||||
retryPolicy: NETWORK_RETRY,
|
||||
handler: (t) =>
|
||||
t.step.run("snapshot", () =>
|
||||
snapshotSite(rt(), { siteId: t.input.siteId, kind: t.input.kind, maxPages: t.input.maxPages }),
|
||||
),
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
type P = Record<string, unknown>;
|
||||
|
||||
export const sourcesOutboxRoutes: Record<string, OutboxRouteFn> = {
|
||||
"feed.due": (ev) => {
|
||||
const p = ev.payload as P;
|
||||
return { task: "ingestFeed", input: { feedSourceId: p.feedId, manual: p.reason === "manual" } };
|
||||
},
|
||||
"release.observed": (ev) => {
|
||||
const p = ev.payload as P;
|
||||
return {
|
||||
task: "resolveObservation",
|
||||
input: { sourceReleaseId: p.sourceReleaseId, observationId: p.observationId },
|
||||
};
|
||||
},
|
||||
"feed.gap_suspected": (ev) => {
|
||||
const p = ev.payload as P;
|
||||
if (!p.siteId) return null;
|
||||
return { task: "snapshotSite", input: { siteId: p.siteId, kind: "backfill", maxPages: BACKFILL_MAX_PAGES } };
|
||||
},
|
||||
};
|
||||
Reference in new issue
Block a user