mirror of
https://github.com/langgenius/dify.git
synced 2026-09-05 00:31:19 +08:00
fix(knowledge_fs): recover source sync status after reload
This commit is contained in:
Stephen Zhou
parent
28da39f909
commit
c7ef31fbeb
@ -1434,6 +1434,9 @@ class KnowledgeFSSourceSyncPolicyResponse(ResponseModel):
|
||||
|
||||
|
||||
class KnowledgeFSSourceResponse(ResponseModel):
|
||||
sync_workflow: KnowledgeFSSourceWorkflowResponse | None = Field(
|
||||
default=None, validation_alias=AliasChoices("sync_workflow", "syncWorkflow")
|
||||
)
|
||||
id: str
|
||||
connection_id: str | None = Field(default=None, validation_alias=AliasChoices("connection_id", "connectionId"))
|
||||
credential_configured: bool | None = Field(
|
||||
|
||||
@ -174,6 +174,26 @@ class RecordingRemote:
|
||||
return {
|
||||
"items": [
|
||||
{
|
||||
"syncWorkflow": {
|
||||
"canceledAt": None,
|
||||
"checkpoint": "provider-read",
|
||||
"completedAt": None,
|
||||
"createdAt": "2030-01-01T00:00:00Z",
|
||||
"cursor": None,
|
||||
"executionAttempts": 1,
|
||||
"id": "workflow-1",
|
||||
"knowledgeSpaceId": "space-1",
|
||||
"kind": "sync",
|
||||
"lastErrorCode": None,
|
||||
"maxExecutionAttempts": 3,
|
||||
"progressCompleted": 2,
|
||||
"progressFailed": 0,
|
||||
"progressSkipped": 0,
|
||||
"progressTotal": 5,
|
||||
"sourceId": "source-1",
|
||||
"state": "syncing",
|
||||
"updatedAt": "2030-01-01T00:01:00Z",
|
||||
},
|
||||
"connectionId": None,
|
||||
"createdAt": "2030-01-01T00:00:00Z",
|
||||
"id": "source-1",
|
||||
@ -977,6 +997,9 @@ def test_basic_product_facade_resolves_control_space_then_uses_exact_kfs_routes(
|
||||
assert updated.settings.embedding is not None
|
||||
assert updated.settings.embedding.plugin_id == "plugin-1"
|
||||
assert len(sources.data) == 1
|
||||
assert sources.data[0].sync_workflow is not None
|
||||
assert sources.data[0].sync_workflow.id == "workflow-1"
|
||||
assert sources.data[0].sync_workflow.state == "syncing"
|
||||
assert sources.data[0].last_synced_at == datetime(2030, 1, 1, 1, tzinfo=UTC)
|
||||
assert sources.data[0].sync_policy is not None
|
||||
assert sources.data[0].sync_policy.mode == "provider"
|
||||
|
||||
@ -0,0 +1,66 @@
|
||||
# Source sync status recovery
|
||||
|
||||
Date: 2026-08-07
|
||||
|
||||
## What changed
|
||||
|
||||
- Enriched source-list responses with the caller-visible latest sync workflow, including terminal
|
||||
outcomes, so the Dify Sources page can recover status after a reload without per-row requests.
|
||||
- Kept the source-list `status` consistent with the latest relevant workflow and normalized the
|
||||
same invariant at the Dify frontend contract boundary, so restored runs participate in Syncing
|
||||
and Error filters even when the source's persisted status has not changed.
|
||||
- Replaced row-owned workflow polling with the source list's bounded bulk snapshot. Terminal
|
||||
failures can no longer disappear in the race between list refresh and workflow-detail polling,
|
||||
and a page with many active sources issues one periodic list request instead of one request per row.
|
||||
- Kept a newly accepted sync authoritative over an older cached terminal run until the source list
|
||||
observes the same run or a newer one, including when the immediate list refresh fails.
|
||||
- Let a newer terminal server run supersede an older local active override, preventing the page from
|
||||
remaining stuck in Syncing when another actor starts and finishes a later run between polls.
|
||||
- Preserved a Source's explicit Disabled state while still returning its latest workflow, and kept
|
||||
newer toggle responses authoritative when a workflow-enriched list replica is stale.
|
||||
- Kept in-flight workflow state across non-enriched toggle responses when the immediate list
|
||||
refresh fails, while still dropping terminal workflows made stale by the source update.
|
||||
- Matched the backend's active-first workflow precedence while reconciling local overrides, so a
|
||||
retried older run cannot be hidden by a newer terminal run retained in the page state.
|
||||
- Bound both permission-scope branches independently for TiDB positional parameters while retaining
|
||||
PostgreSQL placeholder reuse.
|
||||
- Made the in-memory source-workflow repository resolve Capability-owned run scopes explicitly,
|
||||
matching the database repository's fail-closed behavior. Missing Capability provenance no longer
|
||||
behaves like a public empty legacy scope.
|
||||
- Added cross-layer DTO/generated-contract coverage and regressions for restored status, terminal
|
||||
failures, newer runs, status filtering, bulk latest-run lookup, and Capability scope filtering.
|
||||
|
||||
## Why
|
||||
|
||||
The Sources page previously lost an accepted sync run on reload. Returning the latest workflow from
|
||||
the bounded source-list lookup restores both active progress and terminal outcomes. During review,
|
||||
eight follow-up defects were found: a list refresh could remove the run id before detail polling saw
|
||||
its terminal failure; restoring many active rows created a per-row polling fan-out; a restored run
|
||||
retained the source's persisted Active status in list filters; and the in-memory adapter treated a
|
||||
Capability run's intentionally absent legacy scope as public even though the database adapter
|
||||
resolves the immutable Capability grant scope. An older terminal run cached by the source list could
|
||||
also hide a newly accepted sync and stop polling when the immediate refresh failed. Finally, a
|
||||
non-enriched toggle response could discard an in-flight run, and an older retried run could lose to a
|
||||
newer terminal page override even though the backend correctly ranked the active run first. A local
|
||||
active override could also hide a later terminal server run forever when another actor completed a
|
||||
newer sync between list polls.
|
||||
|
||||
## Verification
|
||||
|
||||
- `pnpm --filter @knowledge/api exec vitest run src/source-handlers-coverage.test.ts src/source-product-workflow.test.ts src/source-product-workflow-database-repository.test.ts src/source-product-workflow-memory-repository.test.ts`
|
||||
passed: 141 tests.
|
||||
- `pnpm --filter @knowledge/api typecheck` passed.
|
||||
- Targeted Biome checks passed for all changed KnowledgeFS API files.
|
||||
- `vp test run features/new-rag/__tests__/sources-page.spec.tsx` passed: 41 tests. The three review
|
||||
regressions failed before the stale-override reconciliation fixes and passed afterward.
|
||||
- Targeted `vp check` passed for the changed Sources page, model, and test files.
|
||||
- Targeted KnowledgeFS facade and Swagger/schema tests passed: 131 tests.
|
||||
- Ruff passed for the changed Python DTO and facade test.
|
||||
|
||||
## Risks and follow-up
|
||||
|
||||
- The complete KnowledgeFS `pnpm check`, repository-wide build, and repository-wide lint suites were
|
||||
not rerun because they include unrelated CI, Docker, evaluation, and coverage gates. The affected
|
||||
package tests, typecheck, and targeted lint/build-equivalent checks were run instead.
|
||||
- The full Web `pnpm type-check` remains blocked by pre-existing generated `.next/types/validator.ts`
|
||||
route-type errors. Targeted Vite+ type checking for every changed Web file passed.
|
||||
@ -1394,7 +1394,7 @@ describe("source handlers without optional collaborators", () => {
|
||||
sourceDocumentMaterializer?: SourceDocumentMaterializer;
|
||||
sourceProductWorkflows?: Pick<
|
||||
SourceProductWorkflowRepository,
|
||||
"listLatestSyncCompletions" | "listSyncPolicies"
|
||||
"listLatestSyncCompletions" | "listLatestSyncRuns" | "listSyncPolicies"
|
||||
>;
|
||||
sources?: SourceRepository;
|
||||
websiteCrawlConnector?: WebsiteCrawlConnector;
|
||||
@ -1480,8 +1480,13 @@ describe("source handlers without optional collaborators", () => {
|
||||
it("enriches a source list with product sync details in bulk lookups", async () => {
|
||||
const listSyncPolicies = vi.fn();
|
||||
const listLatestSyncCompletions = vi.fn();
|
||||
const listLatestSyncRuns = vi.fn();
|
||||
const bare = createBareApp({
|
||||
sourceProductWorkflows: { listLatestSyncCompletions, listSyncPolicies },
|
||||
sourceProductWorkflows: {
|
||||
listLatestSyncCompletions,
|
||||
listLatestSyncRuns,
|
||||
listSyncPolicies,
|
||||
},
|
||||
});
|
||||
const { sourceId, spaceId } = await seedSource(bare, "web");
|
||||
listSyncPolicies.mockResolvedValue([
|
||||
@ -1506,6 +1511,29 @@ describe("source handlers without optional collaborators", () => {
|
||||
listLatestSyncCompletions.mockResolvedValue([
|
||||
{ completedAt: "2026-07-14T01:00:00.000Z", sourceId },
|
||||
]);
|
||||
listLatestSyncRuns.mockResolvedValue([
|
||||
{
|
||||
activeSlot: 1,
|
||||
checkpoint: "provider-read",
|
||||
createdAt: "2026-07-14T02:00:00.000Z",
|
||||
executionAttempts: 1,
|
||||
id: "00000000-0000-4000-8000-000000000222",
|
||||
idempotencyKey: "source-sync-active",
|
||||
knowledgeSpaceId: spaceId,
|
||||
kind: "sync",
|
||||
maxExecutionAttempts: 5,
|
||||
payload: {},
|
||||
progressCompleted: 2,
|
||||
progressFailed: 0,
|
||||
progressSkipped: 0,
|
||||
progressTotal: 5,
|
||||
rowVersion: 2,
|
||||
sourceId,
|
||||
state: "syncing",
|
||||
tenantId: "tenant-1",
|
||||
updatedAt: "2026-07-14T02:01:00.000Z",
|
||||
},
|
||||
]);
|
||||
|
||||
const response = await bare.app.request(`/knowledge-spaces/${spaceId}/sources`);
|
||||
|
||||
@ -1513,8 +1541,16 @@ describe("source handlers without optional collaborators", () => {
|
||||
await expect(response.json()).resolves.toMatchObject({
|
||||
items: [
|
||||
{
|
||||
syncWorkflow: {
|
||||
id: "00000000-0000-4000-8000-000000000222",
|
||||
progressCompleted: 2,
|
||||
progressTotal: 5,
|
||||
sourceId,
|
||||
state: "syncing",
|
||||
},
|
||||
id: sourceId,
|
||||
lastSyncedAt: "2026-07-14T01:00:00.000Z",
|
||||
status: "syncing",
|
||||
syncPolicy: {
|
||||
enabled: true,
|
||||
mode: "provider",
|
||||
@ -1535,6 +1571,124 @@ describe("source handlers without optional collaborators", () => {
|
||||
sourceIds: [sourceId],
|
||||
tenantId: "tenant-1",
|
||||
});
|
||||
expect(listLatestSyncRuns).toHaveBeenCalledOnce();
|
||||
expect(listLatestSyncRuns).toHaveBeenCalledWith({
|
||||
candidateGrants: [],
|
||||
knowledgeSpaceId: spaceId,
|
||||
sourceIds: [sourceId],
|
||||
tenantId: "tenant-1",
|
||||
});
|
||||
});
|
||||
|
||||
it("keeps the latest terminal sync result in the source list", async () => {
|
||||
const listLatestSyncRuns = vi.fn();
|
||||
const bare = createBareApp({
|
||||
sourceProductWorkflows: {
|
||||
listLatestSyncCompletions: vi.fn().mockResolvedValue([]),
|
||||
listLatestSyncRuns,
|
||||
listSyncPolicies: vi.fn().mockResolvedValue([]),
|
||||
},
|
||||
});
|
||||
const { sourceId, spaceId } = await seedSource(bare, "web");
|
||||
listLatestSyncRuns.mockResolvedValue([
|
||||
{
|
||||
checkpoint: "provider-read",
|
||||
createdAt: "2030-07-14T02:00:00.000Z",
|
||||
executionAttempts: 1,
|
||||
id: "00000000-0000-4000-8000-000000000333",
|
||||
idempotencyKey: "source-sync-failed",
|
||||
knowledgeSpaceId: spaceId,
|
||||
kind: "sync",
|
||||
lastErrorCode: "PROVIDER_FAILED",
|
||||
maxExecutionAttempts: 5,
|
||||
payload: {},
|
||||
progressCompleted: 2,
|
||||
progressFailed: 1,
|
||||
progressSkipped: 0,
|
||||
progressTotal: 5,
|
||||
rowVersion: 3,
|
||||
sourceId,
|
||||
state: "failed",
|
||||
tenantId: "tenant-1",
|
||||
updatedAt: "2030-07-14T02:01:00.000Z",
|
||||
},
|
||||
]);
|
||||
|
||||
const response = await bare.app.request(`/knowledge-spaces/${spaceId}/sources`);
|
||||
|
||||
expect(response.status).toBe(200);
|
||||
await expect(response.json()).resolves.toMatchObject({
|
||||
items: [
|
||||
{
|
||||
id: sourceId,
|
||||
status: "error",
|
||||
syncWorkflow: {
|
||||
id: "00000000-0000-4000-8000-000000000333",
|
||||
lastErrorCode: "PROVIDER_FAILED",
|
||||
state: "failed",
|
||||
},
|
||||
},
|
||||
],
|
||||
});
|
||||
});
|
||||
|
||||
it("keeps a disabled source disabled while exposing its active sync run", async () => {
|
||||
const listLatestSyncRuns = vi.fn();
|
||||
const bare = createBareApp({
|
||||
sourceProductWorkflows: {
|
||||
listLatestSyncCompletions: vi.fn().mockResolvedValue([]),
|
||||
listLatestSyncRuns,
|
||||
listSyncPolicies: vi.fn().mockResolvedValue([]),
|
||||
},
|
||||
});
|
||||
const { sourceId, spaceId } = await seedSource(bare, "web");
|
||||
const source = await bare.sources.get({ id: sourceId, knowledgeSpaceId: spaceId });
|
||||
expect(source).not.toBeNull();
|
||||
await bare.sources.update({
|
||||
expectedVersion: source?.version,
|
||||
id: sourceId,
|
||||
knowledgeSpaceId: spaceId,
|
||||
status: "disabled",
|
||||
});
|
||||
listLatestSyncRuns.mockResolvedValue([
|
||||
{
|
||||
activeSlot: 1,
|
||||
checkpoint: "provider-read",
|
||||
createdAt: "2030-07-14T02:00:00.000Z",
|
||||
executionAttempts: 1,
|
||||
id: "00000000-0000-4000-8000-000000000444",
|
||||
idempotencyKey: "source-sync-active-disabled",
|
||||
knowledgeSpaceId: spaceId,
|
||||
kind: "sync",
|
||||
maxExecutionAttempts: 5,
|
||||
payload: {},
|
||||
progressCompleted: 2,
|
||||
progressFailed: 0,
|
||||
progressSkipped: 0,
|
||||
progressTotal: 5,
|
||||
rowVersion: 2,
|
||||
sourceId,
|
||||
state: "syncing",
|
||||
tenantId: "tenant-1",
|
||||
updatedAt: "2030-07-14T02:01:00.000Z",
|
||||
},
|
||||
]);
|
||||
|
||||
const response = await bare.app.request(`/knowledge-spaces/${spaceId}/sources`);
|
||||
|
||||
expect(response.status).toBe(200);
|
||||
await expect(response.json()).resolves.toMatchObject({
|
||||
items: [
|
||||
{
|
||||
id: sourceId,
|
||||
status: "disabled",
|
||||
syncWorkflow: {
|
||||
id: "00000000-0000-4000-8000-000000000444",
|
||||
state: "syncing",
|
||||
},
|
||||
},
|
||||
],
|
||||
});
|
||||
});
|
||||
|
||||
it("enforces handler-local space, authorization, and candidate-scope fences", async () => {
|
||||
|
||||
@ -37,7 +37,11 @@ import type {
|
||||
SourceDocumentMaterializer,
|
||||
} from "./source-document-materializer";
|
||||
import { safeSourceOperationError, sourceOperationFailureMetadata } from "./source-operation-error";
|
||||
import type { SourceProductWorkflowRepository } from "./source-product-workflow";
|
||||
import {
|
||||
type SourceProductWorkflowRepository,
|
||||
type SourceWorkflowRun,
|
||||
toPublicSourceWorkflowRun,
|
||||
} from "./source-product-workflow";
|
||||
import {
|
||||
SourceCapacityExceededError,
|
||||
type SourceCursor,
|
||||
@ -76,7 +80,10 @@ export interface RegisterSourceHandlersOptions {
|
||||
readonly sourceCredentials?: SourceCredentialService | undefined;
|
||||
readonly sourceDocumentMaterializer?: SourceDocumentMaterializer | undefined;
|
||||
readonly sourceProductWorkflows?:
|
||||
| Pick<SourceProductWorkflowRepository, "listLatestSyncCompletions" | "listSyncPolicies">
|
||||
| Pick<
|
||||
SourceProductWorkflowRepository,
|
||||
"listLatestSyncCompletions" | "listLatestSyncRuns" | "listSyncPolicies"
|
||||
>
|
||||
| undefined;
|
||||
readonly sources: SourceRepository;
|
||||
readonly spaces: KnowledgeSpaceRepository;
|
||||
@ -227,7 +234,7 @@ export function registerSourceHandlers({
|
||||
}
|
||||
|
||||
const sourceIds = result.items.map((source) => source.id);
|
||||
const [syncPolicies, syncCompletions] = sourceProductWorkflows
|
||||
const [syncPolicies, syncCompletions, latestSyncRuns] = sourceProductWorkflows
|
||||
? await Promise.all([
|
||||
sourceProductWorkflows.listSyncPolicies({
|
||||
knowledgeSpaceId: params.id,
|
||||
@ -239,22 +246,39 @@ export function registerSourceHandlers({
|
||||
sourceIds,
|
||||
tenantId: subject.tenantId,
|
||||
}),
|
||||
sourceProductWorkflows.listLatestSyncRuns({
|
||||
candidateGrants,
|
||||
knowledgeSpaceId: params.id,
|
||||
sourceIds,
|
||||
tenantId: subject.tenantId,
|
||||
}),
|
||||
])
|
||||
: [[], []];
|
||||
: [[], [], []];
|
||||
const syncPoliciesBySourceId = new Map(
|
||||
syncPolicies.map((policy) => [policy.sourceId, SourceSyncPolicyResponseSchema.parse(policy)]),
|
||||
);
|
||||
const lastSyncedAtBySourceId = new Map(
|
||||
syncCompletions.map((completion) => [completion.sourceId, completion.completedAt]),
|
||||
);
|
||||
const latestSyncRunBySourceId = new Map(
|
||||
latestSyncRuns.flatMap((run) => (run.sourceId ? [[run.sourceId, run] as const] : [])),
|
||||
);
|
||||
|
||||
return context.json(
|
||||
{
|
||||
items: result.items.map((source) => {
|
||||
const lastSyncedAt = lastSyncedAtBySourceId.get(source.id);
|
||||
const syncPolicy = syncPoliciesBySourceId.get(source.id);
|
||||
const latestSyncRun = latestSyncRunBySourceId.get(source.id);
|
||||
const syncWorkflow =
|
||||
latestSyncRun &&
|
||||
(latestSyncRun.activeSlot === 1 || latestSyncRun.updatedAt >= source.updatedAt)
|
||||
? toPublicSourceWorkflowRun(latestSyncRun)
|
||||
: undefined;
|
||||
return {
|
||||
...toSourceResponse(source),
|
||||
status: sourceStatusWithSyncWorkflow(source, syncWorkflow ? latestSyncRun : undefined),
|
||||
...(syncWorkflow ? { syncWorkflow } : {}),
|
||||
...(lastSyncedAt ? { lastSyncedAt } : {}),
|
||||
...(syncPolicy ? { syncPolicy } : {}),
|
||||
};
|
||||
@ -1144,6 +1168,13 @@ export function registerSourceHandlers({
|
||||
});
|
||||
}
|
||||
|
||||
function sourceStatusWithSyncWorkflow(source: Source, run?: SourceWorkflowRun): Source["status"] {
|
||||
if (source.status === "disabled" || !run) return source.status;
|
||||
if (run.state === "failed" || run.state === "canceled") return "error";
|
||||
if (run.state === "completed" || run.state === "zero_results") return "active";
|
||||
return "syncing";
|
||||
}
|
||||
|
||||
type SourceRequestContext = Parameters<
|
||||
Parameters<OpenAPIHono<KnowledgeGatewayEnv>["openapi"]>[1]
|
||||
>[0];
|
||||
|
||||
@ -77,6 +77,7 @@ describe.each(["postgres", "tidb"] as const)(
|
||||
tenantId,
|
||||
knowledgeSpaceId,
|
||||
JSON.stringify(["team:camera"]),
|
||||
...(dialect === "tidb" ? [JSON.stringify(["team:camera"])] : []),
|
||||
now,
|
||||
runId,
|
||||
6,
|
||||
@ -1838,6 +1839,92 @@ describe("database source-product workflow repository edge coverage", () => {
|
||||
expect(calls).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("lists the latest sync run for each requested source", async () => {
|
||||
const calls: DatabaseExecuteInput[] = [];
|
||||
const database = testDatabase("postgres", async (input) => {
|
||||
calls.push(input);
|
||||
return {
|
||||
rows: [
|
||||
sourceRunRow("running"),
|
||||
{
|
||||
...sourceRunRow("running"),
|
||||
id: "00000000-0000-4000-8000-000000000222",
|
||||
source_id: "source-b",
|
||||
},
|
||||
],
|
||||
rowsAffected: 2,
|
||||
};
|
||||
});
|
||||
const repository = createDatabaseSourceProductWorkflowRepository({ database });
|
||||
|
||||
await expect(
|
||||
repository.listLatestSyncRuns({
|
||||
candidateGrants: ["team:reader"],
|
||||
knowledgeSpaceId,
|
||||
sourceIds: [sourceId, "source-b", sourceId],
|
||||
tenantId,
|
||||
}),
|
||||
).resolves.toEqual([
|
||||
expect.objectContaining({ sourceId, state: "running" }),
|
||||
expect.objectContaining({ sourceId: "source-b", state: "running" }),
|
||||
]);
|
||||
|
||||
expect(calls).toHaveLength(1);
|
||||
expect(calls[0]).toMatchObject({
|
||||
maxRows: 2,
|
||||
operation: "select",
|
||||
params: [tenantId, knowledgeSpaceId, '["team:reader"]', "sync", sourceId, "source-b"],
|
||||
tableName: "source_workflow_runs",
|
||||
});
|
||||
expect(calls[0]?.sql).toContain("ROW_NUMBER() OVER");
|
||||
expect(calls[0]?.sql).toContain('CASE WHEN "active_slot" = 1 THEN 1 ELSE 0 END DESC');
|
||||
expect(calls[0]?.sql).toContain('"source_run_rank" = 1');
|
||||
expect(calls[0]?.sql).toContain('"source_id" IN ($5, $6)');
|
||||
expect(calls[0]?.sql).toContain('"required_permission_scope"');
|
||||
expect(calls[0]?.sql).toContain('"capability_grants"');
|
||||
await expect(
|
||||
repository.listLatestSyncRuns({
|
||||
candidateGrants: [],
|
||||
knowledgeSpaceId,
|
||||
sourceIds: [],
|
||||
tenantId,
|
||||
}),
|
||||
).resolves.toEqual([]);
|
||||
expect(calls).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("binds both permission-scope branches for TiDB latest sync runs", async () => {
|
||||
const calls: DatabaseExecuteInput[] = [];
|
||||
const database = testDatabase("tidb", async (input) => {
|
||||
calls.push(input);
|
||||
return empty();
|
||||
});
|
||||
const repository = createDatabaseSourceProductWorkflowRepository({ database });
|
||||
|
||||
await expect(
|
||||
repository.listLatestSyncRuns({
|
||||
candidateGrants: ["team:reader"],
|
||||
knowledgeSpaceId,
|
||||
sourceIds: [sourceId],
|
||||
tenantId,
|
||||
}),
|
||||
).resolves.toEqual([]);
|
||||
|
||||
expect(calls).toHaveLength(1);
|
||||
expect(calls[0]?.params).toEqual([
|
||||
tenantId,
|
||||
knowledgeSpaceId,
|
||||
'["team:reader"]',
|
||||
'["team:reader"]',
|
||||
"sync",
|
||||
sourceId,
|
||||
]);
|
||||
const call = calls[0];
|
||||
expect(call).toBeDefined();
|
||||
if (!call) throw new Error("Expected latest sync run query");
|
||||
expect(call.sql.match(/\?/gu)).toHaveLength(call.params.length);
|
||||
});
|
||||
|
||||
it("persists an exact Capability source-policy binding without legacy ACL provenance", async () => {
|
||||
const calls: DatabaseExecuteInput[] = [];
|
||||
const database = testDatabase("postgres", async (input) => {
|
||||
|
||||
@ -222,12 +222,8 @@ export function createDatabaseSourceProductWorkflowRepository(input: {
|
||||
listRuns: async ({ candidateGrants, cursor, knowledgeSpaceId, limit, sourceId, tenantId }) => {
|
||||
listLimit(limit);
|
||||
const readLimit = limit + 1;
|
||||
const params: DatabaseQueryValue[] = [
|
||||
tenantId,
|
||||
knowledgeSpaceId,
|
||||
JSON.stringify(candidateGrants),
|
||||
];
|
||||
let predicate = ` AND ${sourceRunPermissionScopeSql(database, p(database, 3))}`;
|
||||
const params: DatabaseQueryValue[] = [tenantId, knowledgeSpaceId];
|
||||
let predicate = ` AND ${boundSourceRunPermissionScopeSql(database, params, candidateGrants)}`;
|
||||
if (sourceId) {
|
||||
params.push(sourceId);
|
||||
predicate += ` AND ${q(database, "source_id")} = ${p(database, params.length)}`;
|
||||
@ -249,12 +245,8 @@ export function createDatabaseSourceProductWorkflowRepository(input: {
|
||||
listRecentRuns: async ({ candidateGrants, cursor, knowledgeSpaceId, limit, tenantId }) => {
|
||||
listLimit(limit);
|
||||
const readLimit = limit + 1;
|
||||
const params: DatabaseQueryValue[] = [
|
||||
tenantId,
|
||||
knowledgeSpaceId,
|
||||
JSON.stringify(candidateGrants),
|
||||
];
|
||||
let predicate = ` AND ${sourceRunPermissionScopeSql(database, p(database, 3))}`;
|
||||
const params: DatabaseQueryValue[] = [tenantId, knowledgeSpaceId];
|
||||
let predicate = ` AND ${boundSourceRunPermissionScopeSql(database, params, candidateGrants)}`;
|
||||
if (cursor) {
|
||||
params.push(cursor.createdAt, cursor.id);
|
||||
predicate += ` AND (${q(database, "created_at")} < ${p(
|
||||
@ -1152,6 +1144,34 @@ export function createDatabaseSourceProductWorkflowRepository(input: {
|
||||
});
|
||||
return result.rows.map(mapPolicy);
|
||||
},
|
||||
listLatestSyncRuns: async ({ candidateGrants, knowledgeSpaceId, sourceIds, tenantId }) => {
|
||||
const ids = validatedSourceBatchIds(sourceIds);
|
||||
if (ids.length === 0) return [];
|
||||
const params: DatabaseQueryValue[] = [tenantId, knowledgeSpaceId];
|
||||
const permissionScopePredicate = boundSourceRunPermissionScopeSql(
|
||||
database,
|
||||
params,
|
||||
candidateGrants,
|
||||
);
|
||||
params.push("sync");
|
||||
const kindPlaceholder = p(database, params.length);
|
||||
const placeholders = ids
|
||||
.map((id) => {
|
||||
params.push(id);
|
||||
return p(database, params.length);
|
||||
})
|
||||
.join(", ");
|
||||
const rankColumn = "source_run_rank";
|
||||
const rankedRuns = "ranked_source_workflow_runs";
|
||||
const result = await database.execute({
|
||||
maxRows: ids.length,
|
||||
operation: "select",
|
||||
params,
|
||||
sql: `SELECT * FROM (SELECT ${q(database, runTable)}.*, ROW_NUMBER() OVER (PARTITION BY ${q(database, "source_id")} ORDER BY CASE WHEN ${q(database, "active_slot")} = 1 THEN 1 ELSE 0 END DESC, ${q(database, "created_at")} DESC, ${q(database, "updated_at")} DESC, ${q(database, "id")} DESC) AS ${q(database, rankColumn)} FROM ${q(database, runTable)} WHERE ${q(database, "tenant_id")} = ${p(database, 1)} AND ${q(database, "knowledge_space_id")} = ${p(database, 2)} AND ${permissionScopePredicate} AND ${q(database, "kind")} = ${kindPlaceholder} AND ${q(database, "source_id")} IN (${placeholders})) ${q(database, rankedRuns)} WHERE ${q(database, rankColumn)} = 1;`,
|
||||
tableName: runTable,
|
||||
});
|
||||
return result.rows.map(mapRun);
|
||||
},
|
||||
listLatestSyncCompletions: async ({ knowledgeSpaceId, sourceIds, tenantId }) => {
|
||||
const ids = validatedSourceBatchIds(sourceIds);
|
||||
if (ids.length === 0) return [];
|
||||
@ -2435,7 +2455,26 @@ function permissionScopeSql(database: DatabaseAdapter, column: string, grants: s
|
||||
* frozen scope through capability_grants instead of treating the nullable legacy snapshot column
|
||||
* as public content.
|
||||
*/
|
||||
function sourceRunPermissionScopeSql(database: DatabaseAdapter, grants: string): string {
|
||||
function boundSourceRunPermissionScopeSql(
|
||||
database: DatabaseAdapter,
|
||||
params: DatabaseQueryValue[],
|
||||
candidateGrants: readonly string[],
|
||||
): string {
|
||||
const serializedGrants = JSON.stringify(candidateGrants);
|
||||
params.push(serializedGrants);
|
||||
const requiredScopeGrants = p(database, params.length);
|
||||
if (database.dialect === "postgres") {
|
||||
return sourceRunPermissionScopeSql(database, requiredScopeGrants, requiredScopeGrants);
|
||||
}
|
||||
params.push(serializedGrants);
|
||||
return sourceRunPermissionScopeSql(database, requiredScopeGrants, p(database, params.length));
|
||||
}
|
||||
|
||||
function sourceRunPermissionScopeSql(
|
||||
database: DatabaseAdapter,
|
||||
requiredScopeGrants: string,
|
||||
capabilityScopeGrants: string,
|
||||
): string {
|
||||
const runs = q(database, runTable);
|
||||
const provenance = q(database, "source_run_capability");
|
||||
const capabilityGrants = q(database, "capability_grants");
|
||||
@ -2444,7 +2483,7 @@ function sourceRunPermissionScopeSql(database: DatabaseAdapter, grants: string):
|
||||
return `((${requiredScope} IS NOT NULL AND ${permissionScopeSql(
|
||||
database,
|
||||
requiredScope,
|
||||
grants,
|
||||
requiredScopeGrants,
|
||||
)}) OR (${requiredScope} IS NULL AND ${capabilityGrantId} IS NOT NULL AND EXISTS (SELECT 1 FROM ${capabilityGrants} ${provenance} WHERE ${provenance}.${q(
|
||||
database,
|
||||
"tenant_id",
|
||||
@ -2457,7 +2496,7 @@ function sourceRunPermissionScopeSql(database: DatabaseAdapter, grants: string):
|
||||
)} = ${capabilityGrantId} AND ${permissionScopeSql(
|
||||
database,
|
||||
`${provenance}.${q(database, "content_scope_ids")}`,
|
||||
grants,
|
||||
capabilityScopeGrants,
|
||||
)})))`;
|
||||
}
|
||||
function fenceConflict(): never {
|
||||
|
||||
@ -1158,6 +1158,94 @@ describe("in-memory source product workflow repository", () => {
|
||||
}),
|
||||
).resolves.toEqual([]);
|
||||
});
|
||||
|
||||
it("lists the latest sync runs in the requested tenant and space", async () => {
|
||||
const repository = createInMemorySourceProductWorkflowRepository({
|
||||
resolveCapabilityGrantScope: ({
|
||||
grantId,
|
||||
knowledgeSpaceId: scopedSpaceId,
|
||||
tenantId: scopedTenantId,
|
||||
}) =>
|
||||
grantId === "capability-private" &&
|
||||
scopedSpaceId === knowledgeSpaceId &&
|
||||
scopedTenantId === tenantId
|
||||
? ["team:private"]
|
||||
: null,
|
||||
});
|
||||
await repository.start(runRecord("active-sync"));
|
||||
await repository.start(
|
||||
runRecord("other-source", {
|
||||
sourceId: "source-other",
|
||||
}),
|
||||
);
|
||||
await repository.start(
|
||||
runRecord("private-source", {
|
||||
requiredPermissionScope: ["team:private"],
|
||||
sourceId: "source-private",
|
||||
}),
|
||||
);
|
||||
await repository.start(
|
||||
runRecord("capability-private-source", {
|
||||
accessChannel: undefined,
|
||||
capabilityGrantId: "capability-private",
|
||||
permissionSnapshotId: undefined,
|
||||
permissionSnapshotRevision: undefined,
|
||||
requestedBySubjectId: undefined,
|
||||
requiredPermissionScope: undefined,
|
||||
sourceId: "source-capability-private",
|
||||
}),
|
||||
);
|
||||
await terminalRun(repository, "completed-sync", "completed");
|
||||
|
||||
await expect(
|
||||
repository.listLatestSyncRuns({
|
||||
candidateGrants: [],
|
||||
knowledgeSpaceId,
|
||||
sourceIds: [
|
||||
"source-memory",
|
||||
"source-other",
|
||||
"source-private",
|
||||
"source-capability-private",
|
||||
"source-memory",
|
||||
],
|
||||
tenantId,
|
||||
}),
|
||||
).resolves.toEqual([
|
||||
expect.objectContaining({ id: "active-sync", sourceId: "source-memory" }),
|
||||
expect.objectContaining({ id: "other-source", sourceId: "source-other" }),
|
||||
]);
|
||||
await expect(
|
||||
repository.listLatestSyncRuns({
|
||||
candidateGrants: ["team:private"],
|
||||
knowledgeSpaceId,
|
||||
sourceIds: ["source-private"],
|
||||
tenantId,
|
||||
}),
|
||||
).resolves.toEqual([
|
||||
expect.objectContaining({ id: "private-source", sourceId: "source-private" }),
|
||||
]);
|
||||
await expect(
|
||||
repository.listLatestSyncRuns({
|
||||
candidateGrants: ["team:private"],
|
||||
knowledgeSpaceId,
|
||||
sourceIds: ["source-capability-private"],
|
||||
tenantId,
|
||||
}),
|
||||
).resolves.toEqual([
|
||||
expect.objectContaining({
|
||||
id: "capability-private-source",
|
||||
sourceId: "source-capability-private",
|
||||
}),
|
||||
]);
|
||||
await expect(
|
||||
repository.listLatestSyncRuns({
|
||||
candidateGrants: [],
|
||||
knowledgeSpaceId,
|
||||
sourceIds: ["source-memory"],
|
||||
tenantId: "other-tenant",
|
||||
}),
|
||||
).resolves.toEqual([]);
|
||||
});
|
||||
});
|
||||
|
||||
function runRecord(id: string, patch: Partial<NewSourceWorkflowRun> = {}): NewSourceWorkflowRun {
|
||||
|
||||
@ -18,6 +18,13 @@ import {
|
||||
|
||||
export function createInMemorySourceProductWorkflowRepository(input?: {
|
||||
readonly generateLeaseToken?: (() => string) | undefined;
|
||||
readonly resolveCapabilityGrantScope?:
|
||||
| ((input: {
|
||||
readonly grantId: string;
|
||||
readonly knowledgeSpaceId: string;
|
||||
readonly tenantId: string;
|
||||
}) => readonly string[] | null)
|
||||
| undefined;
|
||||
}): SourceProductWorkflowRepository {
|
||||
const generateLeaseToken = input?.generateLeaseToken ?? randomUUID;
|
||||
const runs = new Map<string, SourceWorkflowRun>();
|
||||
@ -27,6 +34,24 @@ export function createInMemorySourceProductWorkflowRepository(input?: {
|
||||
const claimableAt = new Map<string, string>();
|
||||
const policies = new Map<string, SourceSyncPolicyRecord>();
|
||||
const selections = new Map<string, { idempotencyKey: string; pageIds: readonly string[] }>();
|
||||
const runPermissionScopeAllows = (
|
||||
run: SourceWorkflowRun,
|
||||
candidateGrants: readonly string[],
|
||||
): boolean => {
|
||||
if (!run.capabilityGrantId) {
|
||||
return candidatePermissionScopeAllows(run.requiredPermissionScope, candidateGrants);
|
||||
}
|
||||
const requiredScope = input?.resolveCapabilityGrantScope?.({
|
||||
grantId: run.capabilityGrantId,
|
||||
knowledgeSpaceId: run.knowledgeSpaceId,
|
||||
tenantId: run.tenantId,
|
||||
});
|
||||
return (
|
||||
requiredScope !== undefined &&
|
||||
requiredScope !== null &&
|
||||
candidatePermissionScopeAllows(requiredScope, candidateGrants)
|
||||
);
|
||||
};
|
||||
|
||||
const requiredRun = (runId: string) => {
|
||||
const run = runs.get(runId);
|
||||
@ -477,9 +502,7 @@ export function createInMemorySourceProductWorkflowRepository(input?: {
|
||||
listRuns: async ({ candidateGrants, cursor, knowledgeSpaceId, limit, sourceId, tenantId }) => {
|
||||
const list = Array.from(runs.values())
|
||||
.filter((run) => run.tenantId === tenantId && run.knowledgeSpaceId === knowledgeSpaceId)
|
||||
.filter((run) =>
|
||||
candidatePermissionScopeAllows(run.requiredPermissionScope, candidateGrants),
|
||||
)
|
||||
.filter((run) => runPermissionScopeAllows(run, candidateGrants))
|
||||
.filter((run) => !sourceId || run.sourceId === sourceId)
|
||||
.filter((run) => !cursor || run.id > cursor)
|
||||
.sort((left, right) => left.id.localeCompare(right.id));
|
||||
@ -491,9 +514,7 @@ export function createInMemorySourceProductWorkflowRepository(input?: {
|
||||
}
|
||||
const list = Array.from(runs.values())
|
||||
.filter((run) => run.tenantId === tenantId && run.knowledgeSpaceId === knowledgeSpaceId)
|
||||
.filter((run) =>
|
||||
candidatePermissionScopeAllows(run.requiredPermissionScope, candidateGrants),
|
||||
)
|
||||
.filter((run) => runPermissionScopeAllows(run, candidateGrants))
|
||||
.filter(
|
||||
(run) =>
|
||||
!cursor ||
|
||||
@ -636,6 +657,38 @@ export function createInMemorySourceProductWorkflowRepository(input?: {
|
||||
return policy ? [clonePolicy(policy)] : [];
|
||||
});
|
||||
},
|
||||
listLatestSyncRuns: async ({ candidateGrants, knowledgeSpaceId, sourceIds, tenantId }) => {
|
||||
const ids = validatedSourceBatchIds(sourceIds);
|
||||
const requestedSourceIds = new Set(ids);
|
||||
const latestBySourceId = new Map<string, SourceWorkflowRun>();
|
||||
for (const run of runs.values()) {
|
||||
if (
|
||||
run.tenantId !== tenantId ||
|
||||
run.knowledgeSpaceId !== knowledgeSpaceId ||
|
||||
!runPermissionScopeAllows(run, candidateGrants) ||
|
||||
run.kind !== "sync" ||
|
||||
!run.sourceId ||
|
||||
!requestedSourceIds.has(run.sourceId)
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
const latest = latestBySourceId.get(run.sourceId);
|
||||
const runIsActive = run.activeSlot === 1;
|
||||
const latestIsActive = latest?.activeSlot === 1;
|
||||
if (
|
||||
!latest ||
|
||||
(runIsActive && !latestIsActive) ||
|
||||
(runIsActive === latestIsActive &&
|
||||
(run.createdAt > latest.createdAt ||
|
||||
(run.createdAt === latest.createdAt &&
|
||||
(run.updatedAt > latest.updatedAt ||
|
||||
(run.updatedAt === latest.updatedAt && run.id > latest.id)))))
|
||||
) {
|
||||
latestBySourceId.set(run.sourceId, run);
|
||||
}
|
||||
}
|
||||
return [...latestBySourceId.values()].map(cloneRun);
|
||||
},
|
||||
listLatestSyncCompletions: async ({ knowledgeSpaceId, sourceIds, tenantId }) => {
|
||||
const ids = validatedSourceBatchIds(sourceIds);
|
||||
const latestBySourceId = new Map<string, string>();
|
||||
|
||||
@ -877,6 +877,8 @@ describe("source product workflow service boundaries", () => {
|
||||
it("uses capability authorization without minting durable permission snapshots", async () => {
|
||||
const repository = createInMemorySourceProductWorkflowRepository({
|
||||
generateLeaseToken: () => "capability-lease",
|
||||
resolveCapabilityGrantScope: ({ grantId }) =>
|
||||
grantId === "capability-a" ? ["grant:capability"] : null,
|
||||
});
|
||||
const access = accessFixture();
|
||||
const authorization = { authorize: vi.fn(async () => Promise.reject(new Error("unexpected"))) };
|
||||
|
||||
@ -400,6 +400,12 @@ export interface SourceProductWorkflowRepository {
|
||||
readonly sourceIds: readonly string[];
|
||||
readonly tenantId: string;
|
||||
}): Promise<readonly SourceSyncPolicyRecord[]>;
|
||||
listLatestSyncRuns(input: {
|
||||
readonly candidateGrants: readonly string[];
|
||||
readonly knowledgeSpaceId: string;
|
||||
readonly sourceIds: readonly string[];
|
||||
readonly tenantId: string;
|
||||
}): Promise<readonly SourceWorkflowRun[]>;
|
||||
listLatestSyncCompletions(input: {
|
||||
readonly knowledgeSpaceId: string;
|
||||
readonly sourceIds: readonly string[];
|
||||
|
||||
@ -58,6 +58,10 @@ export const SourceWorkflowRunResponseSchema = z
|
||||
})
|
||||
.openapi("SourceWorkflowRun");
|
||||
|
||||
const SourceListItemResponseSchema = SourceResponseSchema.extend({
|
||||
syncWorkflow: SourceWorkflowRunResponseSchema.optional(),
|
||||
});
|
||||
|
||||
export const createSourceRoute = createRoute({
|
||||
method: "post",
|
||||
operationId: "createKnowledgeSpaceSource",
|
||||
@ -105,7 +109,7 @@ export const listSourcesRoute = createRoute({
|
||||
content: {
|
||||
"application/json": {
|
||||
schema: z.object({
|
||||
items: z.array(SourceResponseSchema),
|
||||
items: z.array(SourceListItemResponseSchema),
|
||||
nextCursor: z.string().optional(),
|
||||
}),
|
||||
},
|
||||
|
||||
@ -738,6 +738,7 @@ export type KnowledgeFsSourceResponse = {
|
||||
permission_scope: Array<string>
|
||||
status: 'active' | 'disabled' | 'error' | 'syncing'
|
||||
sync_policy?: KnowledgeFsSourceSyncPolicyResponse | null
|
||||
sync_workflow?: KnowledgeFsSourceWorkflowResponse | null
|
||||
type: 'connector' | 'object-storage' | 'upload' | 'web'
|
||||
updated_at: string
|
||||
uri: string
|
||||
|
||||
@ -574,6 +574,7 @@ export const zKnowledgeFsSourceResponse = z.object({
|
||||
permission_scope: z.array(z.string()),
|
||||
status: z.enum(['active', 'disabled', 'error', 'syncing']),
|
||||
sync_policy: zKnowledgeFsSourceSyncPolicyResponse.nullish(),
|
||||
sync_workflow: zKnowledgeFsSourceWorkflowResponse.nullish(),
|
||||
type: z.enum(['connector', 'object-storage', 'upload', 'web']),
|
||||
updated_at: z.iso.datetime(),
|
||||
uri: z.string(),
|
||||
|
||||
@ -1677,6 +1677,7 @@ export type KnowledgeFsSourceResponse = {
|
||||
permission_scope: Array<string>
|
||||
status: 'active' | 'disabled' | 'error' | 'syncing'
|
||||
sync_policy?: KnowledgeFsSourceSyncPolicyResponse | null
|
||||
sync_workflow?: KnowledgeFsSourceWorkflowResponse | null
|
||||
type: 'connector' | 'object-storage' | 'upload' | 'web'
|
||||
updated_at: string
|
||||
uri: string
|
||||
@ -1706,6 +1707,27 @@ export type KnowledgeFsSourceUpdatePayload = {
|
||||
status?: 'active' | 'disabled' | 'error' | 'syncing' | null
|
||||
}
|
||||
|
||||
export type KnowledgeFsSourceWorkflowResponse = {
|
||||
canceled_at?: string | null
|
||||
checkpoint: string
|
||||
completed_at?: string | null
|
||||
created_at: string
|
||||
cursor?: string | null
|
||||
execution_attempts: number
|
||||
id: string
|
||||
kind: string
|
||||
knowledge_space_id: string
|
||||
last_error_code?: string | null
|
||||
max_execution_attempts: number
|
||||
progress_completed: number
|
||||
progress_failed: number
|
||||
progress_skipped: number
|
||||
progress_total?: number | null
|
||||
source_id?: string | null
|
||||
state: string
|
||||
updated_at: string
|
||||
}
|
||||
|
||||
export type KnowledgeFsSourceWorkspacePagesResponse = {
|
||||
pages: Array<KnowledgeFsSourcePageResponse>
|
||||
total?: number | null
|
||||
|
||||
@ -2023,6 +2023,40 @@ export const zKnowledgeFsSourceSyncPolicyResponse = z.object({
|
||||
updated_at: z.iso.datetime(),
|
||||
})
|
||||
|
||||
/**
|
||||
* KnowledgeFSSourceUpdatePayload
|
||||
*/
|
||||
export const zKnowledgeFsSourceUpdatePayload = z.object({
|
||||
expectedVersion: z.int().gte(1).nullish(),
|
||||
metadata: z.record(z.string(), z.unknown()).nullish(),
|
||||
name: z.string().min(1).max(200).nullish(),
|
||||
status: z.enum(['active', 'disabled', 'error', 'syncing']).nullish(),
|
||||
})
|
||||
|
||||
/**
|
||||
* KnowledgeFSSourceWorkflowResponse
|
||||
*/
|
||||
export const zKnowledgeFsSourceWorkflowResponse = z.object({
|
||||
canceled_at: z.iso.datetime().nullish(),
|
||||
checkpoint: z.string(),
|
||||
completed_at: z.iso.datetime().nullish(),
|
||||
created_at: z.iso.datetime(),
|
||||
cursor: z.string().nullish(),
|
||||
execution_attempts: z.int().gte(0),
|
||||
id: z.string(),
|
||||
kind: z.string(),
|
||||
knowledge_space_id: z.string(),
|
||||
last_error_code: z.string().nullish(),
|
||||
max_execution_attempts: z.int().gte(1),
|
||||
progress_completed: z.int().gte(0),
|
||||
progress_failed: z.int().gte(0),
|
||||
progress_skipped: z.int().gte(0),
|
||||
progress_total: z.int().gte(0).nullish(),
|
||||
source_id: z.string().nullish(),
|
||||
state: z.string(),
|
||||
updated_at: z.iso.datetime(),
|
||||
})
|
||||
|
||||
/**
|
||||
* KnowledgeFSSourceResponse
|
||||
*/
|
||||
@ -2038,6 +2072,7 @@ export const zKnowledgeFsSourceResponse = z.object({
|
||||
permission_scope: z.array(z.string()),
|
||||
status: z.enum(['active', 'disabled', 'error', 'syncing']),
|
||||
sync_policy: zKnowledgeFsSourceSyncPolicyResponse.nullish(),
|
||||
sync_workflow: zKnowledgeFsSourceWorkflowResponse.nullish(),
|
||||
type: z.enum(['connector', 'object-storage', 'upload', 'web']),
|
||||
updated_at: z.iso.datetime(),
|
||||
uri: z.string(),
|
||||
@ -2052,16 +2087,6 @@ export const zKnowledgeFsSourceListResponse = z.object({
|
||||
next_cursor: z.string().nullish(),
|
||||
})
|
||||
|
||||
/**
|
||||
* KnowledgeFSSourceUpdatePayload
|
||||
*/
|
||||
export const zKnowledgeFsSourceUpdatePayload = z.object({
|
||||
expectedVersion: z.int().gte(1).nullish(),
|
||||
metadata: z.record(z.string(), z.unknown()).nullish(),
|
||||
name: z.string().min(1).max(200).nullish(),
|
||||
status: z.enum(['active', 'disabled', 'error', 'syncing']).nullish(),
|
||||
})
|
||||
|
||||
/**
|
||||
* KnowledgeFSSourceWorkspacePagesResponse
|
||||
*/
|
||||
|
||||
@ -1,4 +1,4 @@
|
||||
import type { Source, SourceSyncPolicy } from '../source-models'
|
||||
import type { Source, SourceSyncPolicy, SourceWorkflowRun } from '../source-models'
|
||||
import { screen, waitFor, within } from '@testing-library/react'
|
||||
import userEvent from '@testing-library/user-event'
|
||||
import datasetTranslations from '@/i18n/en-US/dataset.json'
|
||||
@ -11,6 +11,28 @@ const permissionState = vi.hoisted(() => ({
|
||||
workspacePermissionKeys: ['dataset.acl.edit', 'dataset.external.connect'],
|
||||
}))
|
||||
const sourceApiResponse = vi.hoisted(() => (source: Source) => ({
|
||||
sync_workflow: source.syncWorkflow
|
||||
? {
|
||||
canceled_at: source.syncWorkflow.canceledAt ?? null,
|
||||
checkpoint: source.syncWorkflow.checkpoint,
|
||||
completed_at: source.syncWorkflow.completedAt ?? null,
|
||||
created_at: source.syncWorkflow.createdAt,
|
||||
cursor: source.syncWorkflow.cursor ?? null,
|
||||
execution_attempts: source.syncWorkflow.executionAttempts,
|
||||
id: source.syncWorkflow.id,
|
||||
knowledge_space_id: source.syncWorkflow.knowledgeSpaceId,
|
||||
kind: source.syncWorkflow.kind,
|
||||
last_error_code: source.syncWorkflow.lastErrorCode ?? null,
|
||||
max_execution_attempts: source.syncWorkflow.maxExecutionAttempts,
|
||||
progress_completed: source.syncWorkflow.progressCompleted,
|
||||
progress_failed: source.syncWorkflow.progressFailed,
|
||||
progress_skipped: source.syncWorkflow.progressSkipped,
|
||||
progress_total: source.syncWorkflow.progressTotal ?? null,
|
||||
source_id: source.syncWorkflow.sourceId ?? null,
|
||||
state: source.syncWorkflow.state,
|
||||
updated_at: source.syncWorkflow.updatedAt,
|
||||
}
|
||||
: null,
|
||||
connection_id: source.connectionId ?? null,
|
||||
created_at: source.createdAt,
|
||||
credential_configured: source.credentialConfigured ?? null,
|
||||
@ -86,29 +108,7 @@ const settingsState = vi.hoisted(() => ({
|
||||
configurationState: 'active' as 'active' | 'setup-required',
|
||||
refetch: vi.fn(),
|
||||
}))
|
||||
const workflowState = vi.hoisted(() => ({
|
||||
data: undefined as
|
||||
| {
|
||||
canceled_at: null
|
||||
checkpoint: string
|
||||
completed_at: null
|
||||
created_at: string
|
||||
execution_attempts: number
|
||||
id: string
|
||||
kind: string
|
||||
knowledge_space_id: string
|
||||
last_error_code: null
|
||||
max_execution_attempts: number
|
||||
progress_completed: number
|
||||
progress_failed: number
|
||||
progress_skipped: number
|
||||
progress_total: number
|
||||
source_id: string
|
||||
state: string
|
||||
updated_at: string
|
||||
}
|
||||
| undefined,
|
||||
}))
|
||||
|
||||
vi.mock('@tanstack/react-query', async (importOriginal) => {
|
||||
const original = await importOriginal<typeof import('@tanstack/react-query')>()
|
||||
return {
|
||||
@ -124,18 +124,15 @@ vi.mock('@tanstack/react-query', async (importOriginal) => {
|
||||
}
|
||||
: undefined,
|
||||
}),
|
||||
useQuery: (options: { queryKey?: unknown[] }) => {
|
||||
if (options.queryKey?.[1] === 'source-workflow') return { data: workflowState.data }
|
||||
return {
|
||||
data: {
|
||||
configuration_state: settingsState.configurationState,
|
||||
embedding: null,
|
||||
retrieval: null,
|
||||
revision: 1,
|
||||
},
|
||||
refetch: settingsState.refetch,
|
||||
}
|
||||
},
|
||||
useQuery: () => ({
|
||||
data: {
|
||||
configuration_state: settingsState.configurationState,
|
||||
embedding: null,
|
||||
retrieval: null,
|
||||
revision: 1,
|
||||
},
|
||||
refetch: settingsState.refetch,
|
||||
}),
|
||||
useQueryClient: () => ({ invalidateQueries: invalidateQueriesMock }),
|
||||
}
|
||||
})
|
||||
@ -173,15 +170,6 @@ vi.mock('@/service/client', () => ({
|
||||
}),
|
||||
},
|
||||
},
|
||||
sourceWorkflows: {
|
||||
byRunId: {
|
||||
get: {
|
||||
queryOptions: ({ input }: { input: { params: { run_id: string } } }) => ({
|
||||
queryKey: ['knowledge-fs', 'source-workflow', input.params.run_id],
|
||||
}),
|
||||
},
|
||||
},
|
||||
},
|
||||
sources: {
|
||||
get: {
|
||||
infiniteOptions: infiniteOptionsMock,
|
||||
@ -228,6 +216,23 @@ const workflow = (state = 'queued') => ({
|
||||
updated_at: '2026-07-20T10:00:00Z',
|
||||
})
|
||||
|
||||
const sourceWorkflow = (state = 'queued'): SourceWorkflowRun => ({
|
||||
checkpoint: 'sync',
|
||||
createdAt: '2026-07-20T10:00:00Z',
|
||||
executionAttempts: 1,
|
||||
id: 'workflow-1',
|
||||
kind: 'sync',
|
||||
knowledgeSpaceId: 'space-1',
|
||||
maxExecutionAttempts: 3,
|
||||
progressCompleted: 0,
|
||||
progressFailed: 0,
|
||||
progressSkipped: 0,
|
||||
progressTotal: 1,
|
||||
sourceId: 'source-1',
|
||||
state,
|
||||
updatedAt: '2026-07-20T10:00:00Z',
|
||||
})
|
||||
|
||||
describe('SourcesPage', () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks()
|
||||
@ -239,7 +244,7 @@ describe('SourcesPage', () => {
|
||||
sourcesQuery.isPending = false
|
||||
clientMock.deleteSource.mockResolvedValue({ status: 'accepted' })
|
||||
clientMock.patchSource.mockResolvedValue(source({}))
|
||||
workflowState.data = undefined
|
||||
invalidateQueriesMock.mockResolvedValue(undefined)
|
||||
clientMock.syncSource.mockResolvedValue(workflow())
|
||||
permissionState.workspacePermissionKeys = ['dataset.acl.edit', 'dataset.external.connect']
|
||||
settingsState.configurationState = 'active'
|
||||
@ -282,7 +287,13 @@ describe('SourcesPage', () => {
|
||||
expect(
|
||||
options.refetchInterval({
|
||||
state: {
|
||||
data: { pages: [{ data: [sourceApiResponse(source({ status: 'active' }))] }] },
|
||||
data: {
|
||||
pages: [
|
||||
{
|
||||
data: [sourceApiResponse(source({ syncWorkflow: sourceWorkflow('completed') }))],
|
||||
},
|
||||
],
|
||||
},
|
||||
},
|
||||
}),
|
||||
).toBe(false)
|
||||
@ -349,7 +360,12 @@ describe('SourcesPage', () => {
|
||||
{
|
||||
items: [
|
||||
source({ id: 'active', name: 'Product documentation', status: 'active' }),
|
||||
source({ id: 'syncing', name: 'API reference', status: 'syncing' }),
|
||||
source({
|
||||
syncWorkflow: { ...sourceWorkflow('syncing'), sourceId: 'syncing' },
|
||||
id: 'syncing',
|
||||
name: 'API reference',
|
||||
status: 'active',
|
||||
}),
|
||||
source({ id: 'disabled', name: 'Legacy FAQ', status: 'disabled' }),
|
||||
source({ id: 'error', name: 'Support site', status: 'error' }),
|
||||
],
|
||||
@ -374,6 +390,13 @@ describe('SourcesPage', () => {
|
||||
expect(screen.getByText('Support site')).toBeInTheDocument()
|
||||
expect(screen.queryByText('Product documentation')).not.toBeInTheDocument()
|
||||
|
||||
await user.click(sourceFilter)
|
||||
await user.click(
|
||||
screen.getByRole('option', { name: 'dataset.newKnowledge.sourceStatus.syncing' }),
|
||||
)
|
||||
expect(screen.getByText('API reference')).toBeInTheDocument()
|
||||
expect(screen.queryByText('Product documentation')).not.toBeInTheDocument()
|
||||
|
||||
await user.click(sourceFilter)
|
||||
await user.click(screen.getByRole('option', { name: 'dataset.newKnowledge.allSources' }))
|
||||
await user.type(
|
||||
@ -642,7 +665,9 @@ describe('SourcesPage', () => {
|
||||
),
|
||||
).toBeInTheDocument()
|
||||
finishRefresh?.()
|
||||
workflowState.data = workflow('completed')
|
||||
sourcesQuery.data = {
|
||||
pages: [{ items: [source({ syncWorkflow: sourceWorkflow('completed') })] }],
|
||||
}
|
||||
rerender(<SourcesPage knowledgeSpaceId="space-1" />)
|
||||
await waitFor(() =>
|
||||
expect(
|
||||
@ -651,18 +676,491 @@ describe('SourcesPage', () => {
|
||||
),
|
||||
).toBeInTheDocument(),
|
||||
)
|
||||
await waitFor(() => expect(invalidateQueriesMock).toHaveBeenCalledTimes(2))
|
||||
await waitFor(() => expect(invalidateQueriesMock).toHaveBeenCalledTimes(1))
|
||||
const options = infiniteOptionsMock.mock.lastCall?.[0]
|
||||
expect(options).toBeDefined()
|
||||
if (!options) throw new Error('Expected source infinite query options')
|
||||
expect(
|
||||
options.refetchInterval({
|
||||
state: {
|
||||
data: { pages: [{ data: [sourceApiResponse(source({ status: 'active' }))] }] },
|
||||
data: {
|
||||
pages: [
|
||||
{
|
||||
data: [sourceApiResponse(source({ syncWorkflow: sourceWorkflow('completed') }))],
|
||||
},
|
||||
],
|
||||
},
|
||||
},
|
||||
}),
|
||||
).toBe(false)
|
||||
expect(invalidateQueriesMock).toHaveBeenLastCalledWith({ queryKey: ['sources'] })
|
||||
expect(invalidateQueriesMock).toHaveBeenLastCalledWith(
|
||||
{ queryKey: ['sources'] },
|
||||
{ throwOnError: true },
|
||||
)
|
||||
})
|
||||
|
||||
it('restores an active source sync from the server on page load', async () => {
|
||||
sourcesQuery.data = {
|
||||
pages: [{ items: [source({ syncWorkflow: sourceWorkflow('syncing') })] }],
|
||||
}
|
||||
const restored = render(<SourcesPage knowledgeSpaceId="space-1" />)
|
||||
|
||||
expect(
|
||||
within(screen.getByRole('row', { name: /Product documentation/ })).getByText(
|
||||
'dataset.newKnowledge.sourceStatus.syncing',
|
||||
),
|
||||
).toBeInTheDocument()
|
||||
|
||||
sourcesQuery.data = {
|
||||
pages: [{ items: [source({ syncWorkflow: sourceWorkflow('completed') })] }],
|
||||
}
|
||||
restored.rerender(<SourcesPage knowledgeSpaceId="space-1" />)
|
||||
|
||||
await waitFor(() =>
|
||||
expect(
|
||||
within(screen.getByRole('row', { name: /Product documentation/ })).getByText(
|
||||
'dataset.newKnowledge.sourceStatus.active',
|
||||
),
|
||||
).toBeInTheDocument(),
|
||||
)
|
||||
})
|
||||
|
||||
it('keeps a disabled source disabled while its sync is active', async () => {
|
||||
const user = userEvent.setup()
|
||||
sourcesQuery.data = {
|
||||
pages: [
|
||||
{
|
||||
items: [
|
||||
source({
|
||||
status: 'disabled',
|
||||
syncWorkflow: sourceWorkflow('syncing'),
|
||||
}),
|
||||
],
|
||||
},
|
||||
],
|
||||
}
|
||||
|
||||
render(<SourcesPage knowledgeSpaceId="space-1" />)
|
||||
|
||||
const row = screen.getByRole('row', { name: /Product documentation/ })
|
||||
expect(within(row).getByText('dataset.newKnowledge.sourceStatus.disabled')).toBeInTheDocument()
|
||||
expect(
|
||||
within(row).queryByText('dataset.newKnowledge.sourceStatus.syncing'),
|
||||
).not.toBeInTheDocument()
|
||||
|
||||
await user.click(
|
||||
within(row).getByRole('button', {
|
||||
name: 'dataset.newKnowledge.sourceActions:{"name":"Product documentation"}',
|
||||
}),
|
||||
)
|
||||
await user.click(screen.getByRole('menuitem', { name: 'dataset.newKnowledge.syncNow' }))
|
||||
|
||||
expect(within(row).getByText('dataset.newKnowledge.sourceStatus.disabled')).toBeInTheDocument()
|
||||
expect(
|
||||
within(row).queryByText('dataset.newKnowledge.sourceStatus.syncing'),
|
||||
).not.toBeInTheDocument()
|
||||
|
||||
const options = infiniteOptionsMock.mock.lastCall?.[0]
|
||||
expect(options).toBeDefined()
|
||||
if (!options) throw new Error('Expected source infinite query options')
|
||||
expect(
|
||||
options.refetchInterval({
|
||||
state: {
|
||||
data: {
|
||||
pages: [
|
||||
{
|
||||
data: [
|
||||
sourceApiResponse(
|
||||
source({ status: 'disabled', syncWorkflow: sourceWorkflow('syncing') }),
|
||||
),
|
||||
],
|
||||
},
|
||||
],
|
||||
},
|
||||
},
|
||||
}),
|
||||
).toBe(3000)
|
||||
expect(
|
||||
options.refetchInterval({
|
||||
state: {
|
||||
data: {
|
||||
pages: [
|
||||
{
|
||||
data: [
|
||||
sourceApiResponse(
|
||||
source({ status: 'disabled', syncWorkflow: sourceWorkflow('completed') }),
|
||||
),
|
||||
],
|
||||
},
|
||||
],
|
||||
},
|
||||
},
|
||||
}),
|
||||
).toBe(false)
|
||||
})
|
||||
|
||||
it('keeps polling an active sync when enabling a disabled source and refresh fails', async () => {
|
||||
const user = userEvent.setup()
|
||||
const activeWorkflow = sourceWorkflow('syncing')
|
||||
sourcesQuery.data = {
|
||||
pages: [
|
||||
{
|
||||
items: [
|
||||
source({
|
||||
status: 'disabled',
|
||||
syncWorkflow: activeWorkflow,
|
||||
}),
|
||||
],
|
||||
},
|
||||
],
|
||||
}
|
||||
clientMock.patchSource.mockResolvedValue(
|
||||
source({
|
||||
status: 'active',
|
||||
updatedAt: '2026-07-20T10:01:00Z',
|
||||
version: 4,
|
||||
}),
|
||||
)
|
||||
invalidateQueriesMock.mockRejectedValueOnce(new Error('Source refresh failed'))
|
||||
|
||||
render(<SourcesPage knowledgeSpaceId="space-1" />)
|
||||
const row = screen.getByRole('row', { name: /Product documentation/ })
|
||||
await user.click(
|
||||
within(row).getByRole('button', {
|
||||
name: 'dataset.newKnowledge.sourceActions:{"name":"Product documentation"}',
|
||||
}),
|
||||
)
|
||||
await user.click(screen.getByRole('menuitem', { name: 'dataset.enable' }))
|
||||
|
||||
await waitFor(() =>
|
||||
expect(
|
||||
within(row).getByText('dataset.newKnowledge.sourceStatus.syncing'),
|
||||
).toBeInTheDocument(),
|
||||
)
|
||||
const options = infiniteOptionsMock.mock.lastCall?.[0]
|
||||
expect(options).toBeDefined()
|
||||
if (!options) throw new Error('Expected source infinite query options')
|
||||
expect(
|
||||
options.refetchInterval({
|
||||
state: {
|
||||
data: {
|
||||
pages: [
|
||||
{
|
||||
data: [
|
||||
sourceApiResponse(source({ status: 'disabled', syncWorkflow: activeWorkflow })),
|
||||
],
|
||||
},
|
||||
],
|
||||
},
|
||||
},
|
||||
}),
|
||||
).toBe(3000)
|
||||
})
|
||||
|
||||
it('keeps a failed restored sync visible when the source list reaches a terminal state', async () => {
|
||||
sourcesQuery.data = {
|
||||
pages: [{ items: [source({ syncWorkflow: sourceWorkflow('syncing') })] }],
|
||||
}
|
||||
const restored = render(<SourcesPage knowledgeSpaceId="space-1" />)
|
||||
|
||||
sourcesQuery.data = {
|
||||
pages: [
|
||||
{
|
||||
items: [
|
||||
source({
|
||||
syncWorkflow: {
|
||||
...sourceWorkflow('failed'),
|
||||
lastErrorCode: 'PROVIDER_FAILED',
|
||||
},
|
||||
}),
|
||||
],
|
||||
},
|
||||
],
|
||||
}
|
||||
restored.rerender(<SourcesPage knowledgeSpaceId="space-1" />)
|
||||
|
||||
await waitFor(() =>
|
||||
expect(
|
||||
within(screen.getByRole('row', { name: /Product documentation/ })).getByText(
|
||||
'PROVIDER_FAILED',
|
||||
),
|
||||
).toBeInTheDocument(),
|
||||
)
|
||||
expect(invalidateQueriesMock).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('reconciles the source list after a sync finishes', async () => {
|
||||
const user = userEvent.setup()
|
||||
sourcesQuery.data = { pages: [{ items: [source({})] }] }
|
||||
invalidateQueriesMock.mockImplementationOnce((_filters, options) =>
|
||||
options?.throwOnError
|
||||
? Promise.reject(new Error('Initial source refresh failed'))
|
||||
: Promise.resolve(),
|
||||
)
|
||||
|
||||
const rendered = render(<SourcesPage knowledgeSpaceId="space-1" />)
|
||||
await user.click(
|
||||
screen.getByRole('button', {
|
||||
name: 'dataset.newKnowledge.sourceActions:{"name":"Product documentation"}',
|
||||
}),
|
||||
)
|
||||
await user.click(screen.getByRole('menuitem', { name: 'dataset.newKnowledge.syncNow' }))
|
||||
|
||||
expect(
|
||||
within(screen.getByRole('row', { name: /Product documentation/ })).getByText(
|
||||
'dataset.newKnowledge.sourceStatus.syncing',
|
||||
),
|
||||
).toBeInTheDocument()
|
||||
|
||||
sourcesQuery.data = {
|
||||
pages: [{ items: [source({ syncWorkflow: sourceWorkflow('completed') })] }],
|
||||
}
|
||||
rendered.rerender(<SourcesPage knowledgeSpaceId="space-1" />)
|
||||
|
||||
await waitFor(() => expect(invalidateQueriesMock).toHaveBeenCalledTimes(1))
|
||||
await waitFor(() => {
|
||||
const options = infiniteOptionsMock.mock.lastCall?.[0]
|
||||
expect(options).toBeDefined()
|
||||
if (!options) throw new Error('Expected source infinite query options')
|
||||
expect(
|
||||
options.refetchInterval({
|
||||
state: {
|
||||
data: {
|
||||
pages: [
|
||||
{
|
||||
data: [sourceApiResponse(source({ syncWorkflow: sourceWorkflow('completed') }))],
|
||||
},
|
||||
],
|
||||
},
|
||||
},
|
||||
}),
|
||||
).toBe(false)
|
||||
})
|
||||
expect(
|
||||
within(screen.getByRole('row', { name: /Product documentation/ })).getByText(
|
||||
'dataset.newKnowledge.sourceStatus.active',
|
||||
),
|
||||
).toBeInTheDocument()
|
||||
})
|
||||
|
||||
it('tracks a newer server workflow after a local sync completes', async () => {
|
||||
const user = userEvent.setup()
|
||||
sourcesQuery.data = { pages: [{ items: [source({})] }] }
|
||||
|
||||
const rendered = render(<SourcesPage knowledgeSpaceId="space-1" />)
|
||||
await user.click(
|
||||
screen.getByRole('button', {
|
||||
name: 'dataset.newKnowledge.sourceActions:{"name":"Product documentation"}',
|
||||
}),
|
||||
)
|
||||
await user.click(screen.getByRole('menuitem', { name: 'dataset.newKnowledge.syncNow' }))
|
||||
|
||||
sourcesQuery.data = {
|
||||
pages: [{ items: [source({ syncWorkflow: sourceWorkflow('completed') })] }],
|
||||
}
|
||||
rendered.rerender(<SourcesPage knowledgeSpaceId="space-1" />)
|
||||
await waitFor(() =>
|
||||
expect(
|
||||
within(screen.getByRole('row', { name: /Product documentation/ })).getByText(
|
||||
'dataset.newKnowledge.sourceStatus.active',
|
||||
),
|
||||
).toBeInTheDocument(),
|
||||
)
|
||||
|
||||
sourcesQuery.data = {
|
||||
pages: [
|
||||
{
|
||||
items: [
|
||||
source({
|
||||
syncWorkflow: { ...sourceWorkflow('syncing'), id: 'workflow-2' },
|
||||
}),
|
||||
],
|
||||
},
|
||||
],
|
||||
}
|
||||
rendered.rerender(<SourcesPage knowledgeSpaceId="space-1" />)
|
||||
|
||||
await waitFor(() =>
|
||||
expect(
|
||||
within(screen.getByRole('row', { name: /Product documentation/ })).getByText(
|
||||
'dataset.newKnowledge.sourceStatus.syncing',
|
||||
),
|
||||
).toBeInTheDocument(),
|
||||
)
|
||||
})
|
||||
|
||||
it('replaces a local active sync with a newer terminal server workflow', async () => {
|
||||
const user = userEvent.setup()
|
||||
sourcesQuery.data = { pages: [{ items: [source({})] }] }
|
||||
|
||||
const rendered = render(<SourcesPage knowledgeSpaceId="space-1" />)
|
||||
await user.click(
|
||||
screen.getByRole('button', {
|
||||
name: 'dataset.newKnowledge.sourceActions:{"name":"Product documentation"}',
|
||||
}),
|
||||
)
|
||||
await user.click(screen.getByRole('menuitem', { name: 'dataset.newKnowledge.syncNow' }))
|
||||
|
||||
await waitFor(() =>
|
||||
expect(
|
||||
within(screen.getByRole('row', { name: /Product documentation/ })).getByText(
|
||||
'dataset.newKnowledge.sourceStatus.syncing',
|
||||
),
|
||||
).toBeInTheDocument(),
|
||||
)
|
||||
|
||||
sourcesQuery.data = {
|
||||
pages: [
|
||||
{
|
||||
items: [
|
||||
source({
|
||||
syncWorkflow: {
|
||||
...sourceWorkflow('completed'),
|
||||
createdAt: '2026-07-20T11:00:00Z',
|
||||
id: 'workflow-2',
|
||||
updatedAt: '2026-07-20T11:01:00Z',
|
||||
},
|
||||
}),
|
||||
],
|
||||
},
|
||||
],
|
||||
}
|
||||
rendered.rerender(<SourcesPage knowledgeSpaceId="space-1" />)
|
||||
|
||||
await waitFor(() =>
|
||||
expect(
|
||||
within(screen.getByRole('row', { name: /Product documentation/ })).getByText(
|
||||
'dataset.newKnowledge.sourceStatus.active',
|
||||
),
|
||||
).toBeInTheDocument(),
|
||||
)
|
||||
const options = infiniteOptionsMock.mock.lastCall?.[0]
|
||||
expect(options).toBeDefined()
|
||||
if (!options) throw new Error('Expected source infinite query options')
|
||||
expect(
|
||||
options.refetchInterval({
|
||||
state: {
|
||||
data: {
|
||||
pages: [
|
||||
{
|
||||
data: [
|
||||
sourceApiResponse(
|
||||
source({
|
||||
syncWorkflow: {
|
||||
...sourceWorkflow('completed'),
|
||||
createdAt: '2026-07-20T11:00:00Z',
|
||||
id: 'workflow-2',
|
||||
updatedAt: '2026-07-20T11:01:00Z',
|
||||
},
|
||||
}),
|
||||
),
|
||||
],
|
||||
},
|
||||
],
|
||||
},
|
||||
},
|
||||
}),
|
||||
).toBe(false)
|
||||
})
|
||||
|
||||
it('tracks an older workflow when the server has retried it', async () => {
|
||||
const user = userEvent.setup()
|
||||
sourcesQuery.data = { pages: [{ items: [source({})] }] }
|
||||
clientMock.syncSource.mockResolvedValue({
|
||||
...workflow('completed'),
|
||||
created_at: '2026-07-20T11:00:00Z',
|
||||
id: 'newer-terminal-workflow',
|
||||
updated_at: '2026-07-20T11:01:00Z',
|
||||
})
|
||||
|
||||
const rendered = render(<SourcesPage knowledgeSpaceId="space-1" />)
|
||||
await user.click(
|
||||
screen.getByRole('button', {
|
||||
name: 'dataset.newKnowledge.sourceActions:{"name":"Product documentation"}',
|
||||
}),
|
||||
)
|
||||
await user.click(screen.getByRole('menuitem', { name: 'dataset.newKnowledge.syncNow' }))
|
||||
|
||||
const retriedWorkflow = {
|
||||
...sourceWorkflow('syncing'),
|
||||
createdAt: '2026-07-20T09:00:00Z',
|
||||
id: 'older-retried-workflow',
|
||||
updatedAt: '2026-07-20T12:00:00Z',
|
||||
}
|
||||
sourcesQuery.data = {
|
||||
pages: [{ items: [source({ syncWorkflow: retriedWorkflow })] }],
|
||||
}
|
||||
rendered.rerender(<SourcesPage knowledgeSpaceId="space-1" />)
|
||||
|
||||
await waitFor(() =>
|
||||
expect(
|
||||
within(screen.getByRole('row', { name: /Product documentation/ })).getByText(
|
||||
'dataset.newKnowledge.sourceStatus.syncing',
|
||||
),
|
||||
).toBeInTheDocument(),
|
||||
)
|
||||
const options = infiniteOptionsMock.mock.lastCall?.[0]
|
||||
expect(options).toBeDefined()
|
||||
if (!options) throw new Error('Expected source infinite query options')
|
||||
expect(
|
||||
options.refetchInterval({
|
||||
state: {
|
||||
data: {
|
||||
pages: [
|
||||
{
|
||||
data: [sourceApiResponse(source({ syncWorkflow: retriedWorkflow }))],
|
||||
},
|
||||
],
|
||||
},
|
||||
},
|
||||
}),
|
||||
).toBe(3000)
|
||||
})
|
||||
|
||||
it('keeps a newly accepted sync visible when the source list still has an older workflow', async () => {
|
||||
const user = userEvent.setup()
|
||||
sourcesQuery.data = {
|
||||
pages: [{ items: [source({ syncWorkflow: sourceWorkflow('completed') })] }],
|
||||
}
|
||||
clientMock.syncSource.mockResolvedValue({
|
||||
...workflow(),
|
||||
created_at: '2026-07-20T11:00:00Z',
|
||||
id: 'workflow-2',
|
||||
updated_at: '2026-07-20T11:00:00Z',
|
||||
})
|
||||
invalidateQueriesMock.mockRejectedValueOnce(new Error('Source refresh failed'))
|
||||
|
||||
render(<SourcesPage knowledgeSpaceId="space-1" />)
|
||||
await user.click(
|
||||
screen.getByRole('button', {
|
||||
name: 'dataset.newKnowledge.sourceActions:{"name":"Product documentation"}',
|
||||
}),
|
||||
)
|
||||
await user.click(screen.getByRole('menuitem', { name: 'dataset.newKnowledge.syncNow' }))
|
||||
|
||||
await waitFor(() =>
|
||||
expect(
|
||||
within(screen.getByRole('row', { name: /Product documentation/ })).getByText(
|
||||
'dataset.newKnowledge.sourceStatus.syncing',
|
||||
),
|
||||
).toBeInTheDocument(),
|
||||
)
|
||||
const options = infiniteOptionsMock.mock.lastCall?.[0]
|
||||
expect(options).toBeDefined()
|
||||
if (!options) throw new Error('Expected source infinite query options')
|
||||
expect(
|
||||
options.refetchInterval({
|
||||
state: {
|
||||
data: {
|
||||
pages: [
|
||||
{
|
||||
data: [sourceApiResponse(source({ syncWorkflow: sourceWorkflow('completed') }))],
|
||||
},
|
||||
],
|
||||
},
|
||||
},
|
||||
}),
|
||||
).toBe(3000)
|
||||
})
|
||||
|
||||
it('prompts for model setup before syncing a source', async () => {
|
||||
@ -729,7 +1227,7 @@ describe('SourcesPage', () => {
|
||||
)
|
||||
})
|
||||
|
||||
it('uses the returned source version while the list replica is stale', async () => {
|
||||
it('uses the returned source version while a workflow-enriched list replica is stale', async () => {
|
||||
const user = userEvent.setup()
|
||||
const syncPolicy: SourceSyncPolicy = {
|
||||
createdAt: '2026-07-20T10:00:00Z',
|
||||
@ -743,7 +1241,10 @@ describe('SourcesPage', () => {
|
||||
sourceId: 'source-1',
|
||||
updatedAt: '2026-07-20T10:00:00Z',
|
||||
}
|
||||
sourcesQuery.data = { pages: [{ items: [source({ syncPolicy })] }] }
|
||||
sourcesQuery.data = {
|
||||
pages: [{ items: [source({ syncPolicy, syncWorkflow: sourceWorkflow('failed') })] }],
|
||||
}
|
||||
invalidateQueriesMock.mockRejectedValue(new Error('Source refresh failed'))
|
||||
clientMock.patchSource
|
||||
.mockResolvedValueOnce(
|
||||
source({
|
||||
|
||||
@ -20,6 +20,7 @@ export type Source = {
|
||||
name: string
|
||||
permissionScope?: string[]
|
||||
status: 'active' | 'syncing' | 'error' | 'disabled'
|
||||
syncWorkflow?: SourceWorkflowRun
|
||||
syncPolicy?: SourceSyncPolicy
|
||||
type: 'upload' | 'object-storage' | 'connector' | 'web'
|
||||
updatedAt: string
|
||||
@ -110,7 +111,42 @@ export type SourceSyncPolicy = {
|
||||
|
||||
export type SourceSyncPolicyBody = KnowledgeFsSourceSyncPolicyPayload
|
||||
|
||||
const SOURCE_WORKFLOW_SUCCESS_STATES = new Set([
|
||||
'complete',
|
||||
'completed',
|
||||
'success',
|
||||
'succeeded',
|
||||
'zero_results',
|
||||
])
|
||||
const SOURCE_WORKFLOW_FAILURE_STATES = new Set([
|
||||
'canceled',
|
||||
'cancelled',
|
||||
'error',
|
||||
'exhausted',
|
||||
'failed',
|
||||
'timed_out',
|
||||
'timeout',
|
||||
])
|
||||
|
||||
export function sourceWorkflowStatus(state: string): Source['status'] {
|
||||
const normalized = state.trim().toLowerCase().replaceAll('-', '_').replaceAll(' ', '_')
|
||||
if (SOURCE_WORKFLOW_FAILURE_STATES.has(normalized)) return 'error'
|
||||
if (SOURCE_WORKFLOW_SUCCESS_STATES.has(normalized)) return 'active'
|
||||
return 'syncing'
|
||||
}
|
||||
|
||||
export function sourceStatusWithSyncWorkflow(
|
||||
status: Source['status'],
|
||||
syncWorkflow?: SourceWorkflowRun,
|
||||
): Source['status'] {
|
||||
if (status === 'disabled' || !syncWorkflow) return status
|
||||
return sourceWorkflowStatus(syncWorkflow.state)
|
||||
}
|
||||
|
||||
export function sourceFromApi(source: KnowledgeFsSourceResponse): Source {
|
||||
const syncWorkflow = source.sync_workflow
|
||||
? sourceWorkflowFromApi(source.sync_workflow)
|
||||
: undefined
|
||||
return {
|
||||
connectionId: source.connection_id ?? undefined,
|
||||
createdAt: source.created_at,
|
||||
@ -121,7 +157,8 @@ export function sourceFromApi(source: KnowledgeFsSourceResponse): Source {
|
||||
metadata: source.metadata,
|
||||
name: source.name,
|
||||
permissionScope: source.permission_scope,
|
||||
status: source.status,
|
||||
status: sourceStatusWithSyncWorkflow(source.status, syncWorkflow),
|
||||
syncWorkflow,
|
||||
syncPolicy: source.sync_policy ? sourceSyncPolicyFromApi(source.sync_policy) : undefined,
|
||||
type: source.type,
|
||||
updatedAt: source.updated_at,
|
||||
|
||||
@ -33,7 +33,7 @@ import {
|
||||
} from '@langgenius/dify-ui/select'
|
||||
import { StatusDot } from '@langgenius/dify-ui/status-dot'
|
||||
import { toast } from '@langgenius/dify-ui/toast'
|
||||
import { useInfiniteQuery, useQuery, useQueryClient } from '@tanstack/react-query'
|
||||
import { useInfiniteQuery, useQueryClient } from '@tanstack/react-query'
|
||||
import { useAtomValue } from 'jotai'
|
||||
import { useEffect, useMemo, useState } from 'react'
|
||||
import { useTranslation } from 'react-i18next'
|
||||
@ -46,7 +46,12 @@ import { consoleClient, consoleQuery } from '@/service/client'
|
||||
import { hasPermission } from '@/utils/permission'
|
||||
import { KnowledgeModelSetupDialog } from './components/knowledge-model-setup-dialog'
|
||||
import { newKnowledgeAddSourcePath } from './routes'
|
||||
import { sourceFromApi, sourceWorkflowFromApi } from './source-models'
|
||||
import {
|
||||
sourceFromApi,
|
||||
sourceStatusWithSyncWorkflow,
|
||||
sourceWorkflowFromApi,
|
||||
sourceWorkflowStatus,
|
||||
} from './source-models'
|
||||
import { useKnowledgeModelSetupGuard } from './use-knowledge-model-setup-guard'
|
||||
|
||||
type SourceStatus = Source['status']
|
||||
@ -56,23 +61,6 @@ type SourceSort = 'name-asc' | 'name-desc'
|
||||
const PAGE_SIZE = 50
|
||||
const MAX_AUTO_FILTER_PAGES = 4
|
||||
const SOURCE_POLL_INTERVAL = 3000
|
||||
const SOURCE_WORKFLOW_POLL_INTERVAL = 1500
|
||||
const SOURCE_WORKFLOW_SUCCESS_STATES = new Set([
|
||||
'complete',
|
||||
'completed',
|
||||
'success',
|
||||
'succeeded',
|
||||
'zero_results',
|
||||
])
|
||||
const SOURCE_WORKFLOW_FAILURE_STATES = new Set([
|
||||
'canceled',
|
||||
'cancelled',
|
||||
'error',
|
||||
'exhausted',
|
||||
'failed',
|
||||
'timed_out',
|
||||
'timeout',
|
||||
])
|
||||
|
||||
const statusDotStatus: Record<SourceStatus, StatusDotStatus> = {
|
||||
active: 'success',
|
||||
@ -164,17 +152,6 @@ function createIdempotencyKey() {
|
||||
return globalThis.crypto?.randomUUID?.() ?? `${Date.now()}-${Math.random()}`
|
||||
}
|
||||
|
||||
function normalizedWorkflowState(state: string) {
|
||||
return state.trim().toLowerCase().replaceAll('-', '_').replaceAll(' ', '_')
|
||||
}
|
||||
|
||||
function sourceWorkflowStatus(state: string): SourceStatus {
|
||||
const normalized = normalizedWorkflowState(state)
|
||||
if (SOURCE_WORKFLOW_FAILURE_STATES.has(normalized)) return 'error'
|
||||
if (SOURCE_WORKFLOW_SUCCESS_STATES.has(normalized)) return 'active'
|
||||
return 'syncing'
|
||||
}
|
||||
|
||||
function getOpenableSourceUri(uri: string) {
|
||||
try {
|
||||
const url = new URL(uri)
|
||||
@ -190,22 +167,68 @@ function getOpenableSourceUri(uri: string) {
|
||||
}
|
||||
}
|
||||
|
||||
function sourceWorkflowIsActive(workflow?: Source['syncWorkflow']) {
|
||||
return workflow !== undefined && sourceWorkflowStatus(workflow.state) === 'syncing'
|
||||
}
|
||||
|
||||
function latestSourceWorkflow(
|
||||
sourceWorkflow?: Source['syncWorkflow'],
|
||||
sourceOverrideWorkflow?: Source['syncWorkflow'],
|
||||
) {
|
||||
if (!sourceWorkflow || !sourceOverrideWorkflow) return sourceWorkflow ?? sourceOverrideWorkflow
|
||||
if (sourceWorkflow.id === sourceOverrideWorkflow.id) return sourceWorkflow
|
||||
const sourceWorkflowIsRunning = sourceWorkflowIsActive(sourceWorkflow)
|
||||
const sourceOverrideWorkflowIsRunning = sourceWorkflowIsActive(sourceOverrideWorkflow)
|
||||
// The server snapshot ranks active runs first, so an active server run remains authoritative
|
||||
// even when it is an older run being retried. A local active override still has to be newer
|
||||
// than a terminal server run, otherwise it could remain stuck after a later run completes.
|
||||
if (sourceWorkflowIsRunning && !sourceOverrideWorkflowIsRunning) return sourceWorkflow
|
||||
const createdAtComparison = sourceWorkflow.createdAt.localeCompare(
|
||||
sourceOverrideWorkflow.createdAt,
|
||||
)
|
||||
if (createdAtComparison !== 0)
|
||||
return createdAtComparison > 0 ? sourceWorkflow : sourceOverrideWorkflow
|
||||
const updatedAtComparison = sourceWorkflow.updatedAt.localeCompare(
|
||||
sourceOverrideWorkflow.updatedAt,
|
||||
)
|
||||
if (updatedAtComparison !== 0)
|
||||
return updatedAtComparison > 0 ? sourceWorkflow : sourceOverrideWorkflow
|
||||
return sourceWorkflow.id > sourceOverrideWorkflow.id ? sourceWorkflow : sourceOverrideWorkflow
|
||||
}
|
||||
|
||||
function getCurrentSource(source: Source, sourceOverride?: Source) {
|
||||
if (!sourceOverride || sourceOverride.id !== source.id) return source
|
||||
const sourceVersion = source.version ?? -1
|
||||
const overrideVersion = sourceOverride.version ?? -1
|
||||
if (sourceVersion > overrideVersion) return source
|
||||
const currentSource =
|
||||
sourceVersion < overrideVersion || source.updatedAt <= sourceOverride.updatedAt
|
||||
? sourceOverride
|
||||
: source
|
||||
const overrideHasNewerSource =
|
||||
sourceVersion < overrideVersion || source.updatedAt < sourceOverride.updatedAt
|
||||
const sourceHasNewerSource =
|
||||
sourceVersion === overrideVersion && source.updatedAt > sourceOverride.updatedAt
|
||||
if (sourceHasNewerSource) return source
|
||||
const syncWorkflow = overrideHasNewerSource
|
||||
? sourceOverride.syncWorkflow
|
||||
: latestSourceWorkflow(source.syncWorkflow, sourceOverride.syncWorkflow)
|
||||
if (
|
||||
!overrideHasNewerSource &&
|
||||
source.syncWorkflow &&
|
||||
source.syncWorkflow.id !== sourceOverride.syncWorkflow?.id &&
|
||||
syncWorkflow === source.syncWorkflow
|
||||
)
|
||||
return source
|
||||
return {
|
||||
...currentSource,
|
||||
...sourceOverride,
|
||||
lastSyncedAt: source.lastSyncedAt ?? sourceOverride.lastSyncedAt,
|
||||
status: sourceStatusWithSyncWorkflow(sourceOverride.status, syncWorkflow),
|
||||
syncWorkflow,
|
||||
syncPolicy: source.syncPolicy ?? sourceOverride.syncPolicy,
|
||||
}
|
||||
}
|
||||
|
||||
function sourceNeedsPolling(source: Source) {
|
||||
return source.status === 'syncing' || sourceWorkflowIsActive(source.syncWorkflow)
|
||||
}
|
||||
|
||||
type SourceAction = 'remove' | 'sync' | 'toggle'
|
||||
|
||||
function SourceActions({
|
||||
@ -346,7 +369,6 @@ function SourceRow({
|
||||
ensureModelSetupReady,
|
||||
knowledgeSpaceId,
|
||||
onCheckedChange,
|
||||
onSourceReconciled,
|
||||
onRemoved,
|
||||
onSourceChange,
|
||||
source,
|
||||
@ -357,7 +379,6 @@ function SourceRow({
|
||||
ensureModelSetupReady: () => Promise<boolean>
|
||||
knowledgeSpaceId: string
|
||||
onCheckedChange: (checked: boolean) => void
|
||||
onSourceReconciled: () => void
|
||||
onRemoved: () => void
|
||||
onSourceChange: (source: Source) => void
|
||||
source: Source
|
||||
@ -367,40 +388,8 @@ function SourceRow({
|
||||
const { formatTimeFromNow } = useFormatTimeFromNow()
|
||||
const queryClient = useQueryClient()
|
||||
const [pendingAction, setPendingAction] = useState<SourceAction>()
|
||||
const [acceptedSyncRun, setAcceptedSyncRun] = useState<ReturnType<typeof sourceWorkflowFromApi>>()
|
||||
const syncWorkflowQuery = useQuery({
|
||||
...consoleQuery.knowledgeFs.spaces.byControlSpaceId.sourceWorkflows.byRunId.get.queryOptions({
|
||||
input: {
|
||||
params: {
|
||||
control_space_id: knowledgeSpaceId,
|
||||
run_id: acceptedSyncRun?.id ?? '',
|
||||
},
|
||||
},
|
||||
}),
|
||||
enabled: Boolean(acceptedSyncRun),
|
||||
refetchInterval: (query) => {
|
||||
const workflow = query.state.data ? sourceWorkflowFromApi(query.state.data) : acceptedSyncRun
|
||||
return workflow && sourceWorkflowStatus(workflow.state) === 'syncing'
|
||||
? SOURCE_WORKFLOW_POLL_INTERVAL
|
||||
: false
|
||||
},
|
||||
})
|
||||
const syncWorkflow = syncWorkflowQuery.data
|
||||
? sourceWorkflowFromApi(syncWorkflowQuery.data)
|
||||
: acceptedSyncRun
|
||||
const syncWorkflowId = syncWorkflow?.id
|
||||
const syncWorkflowState = syncWorkflow?.state
|
||||
const syncWorkflow = source.syncWorkflow
|
||||
|
||||
useEffect(() => {
|
||||
if (!syncWorkflowState || sourceWorkflowStatus(syncWorkflowState) === 'syncing') return
|
||||
void queryClient.invalidateQueries({
|
||||
queryKey: consoleQuery.knowledgeFs.spaces.byControlSpaceId.sources.get.key(),
|
||||
})
|
||||
}, [queryClient, syncWorkflowId, syncWorkflowState])
|
||||
|
||||
const visibleSource = syncWorkflow
|
||||
? { ...source, status: sourceWorkflowStatus(syncWorkflow.state) }
|
||||
: source
|
||||
const providerName = sourceProviderName(source)
|
||||
const sourceSyncPolicy = source.syncPolicy
|
||||
const syncPolicy = sourceSyncPolicy
|
||||
@ -431,7 +420,6 @@ function SourceRow({
|
||||
action: SourceAction,
|
||||
mutation: () => Promise<Result>,
|
||||
onAccepted?: (result: Result) => void,
|
||||
onRefreshed?: () => void,
|
||||
beforeAction?: () => Promise<boolean>,
|
||||
) => {
|
||||
if (pendingAction) return false
|
||||
@ -455,10 +443,14 @@ function SourceRow({
|
||||
onAccepted?.(result)
|
||||
|
||||
try {
|
||||
await queryClient.invalidateQueries({
|
||||
queryKey: consoleQuery.knowledgeFs.spaces.byControlSpaceId.sources.get.key(),
|
||||
})
|
||||
onRefreshed?.()
|
||||
await queryClient.invalidateQueries(
|
||||
{
|
||||
queryKey: consoleQuery.knowledgeFs.spaces.byControlSpaceId.sources.get.key(),
|
||||
},
|
||||
{
|
||||
throwOnError: true,
|
||||
},
|
||||
)
|
||||
} catch {
|
||||
// The accepted mutation is already reflected by the list-owner state.
|
||||
}
|
||||
@ -478,10 +470,12 @@ function SourceRow({
|
||||
}),
|
||||
(workflow) => {
|
||||
const run = sourceWorkflowFromApi(workflow)
|
||||
setAcceptedSyncRun(run)
|
||||
onSourceChange({ ...source, status: sourceWorkflowStatus(run.state) })
|
||||
onSourceChange({
|
||||
...source,
|
||||
syncWorkflow: run,
|
||||
status: sourceStatusWithSyncWorkflow(source.status, run),
|
||||
})
|
||||
},
|
||||
onSourceReconciled,
|
||||
ensureModelSetupReady,
|
||||
)
|
||||
|
||||
@ -498,12 +492,18 @@ function SourceRow({
|
||||
params: { control_space_id: knowledgeSpaceId, source_id: source.id },
|
||||
}),
|
||||
),
|
||||
(updatedSource) =>
|
||||
(updatedSource) => {
|
||||
const syncWorkflow =
|
||||
updatedSource.syncWorkflow ??
|
||||
(sourceWorkflowIsActive(source.syncWorkflow) ? source.syncWorkflow : undefined)
|
||||
onSourceChange({
|
||||
...updatedSource,
|
||||
lastSyncedAt: updatedSource.lastSyncedAt ?? source.lastSyncedAt,
|
||||
status: sourceStatusWithSyncWorkflow(updatedSource.status, syncWorkflow),
|
||||
syncWorkflow,
|
||||
syncPolicy: updatedSource.syncPolicy ?? source.syncPolicy,
|
||||
}),
|
||||
})
|
||||
},
|
||||
)
|
||||
|
||||
const removeSource = () =>
|
||||
@ -525,7 +525,7 @@ function SourceRow({
|
||||
<tr
|
||||
className={cn(
|
||||
'h-[50px] border-t border-divider-subtle',
|
||||
visibleSource.status === 'disabled' && '[&>td:not(:first-child)]:opacity-60',
|
||||
source.status === 'disabled' && '[&>td:not(:first-child)]:opacity-60',
|
||||
)}
|
||||
>
|
||||
<td className="py-2 pr-3 whitespace-nowrap">
|
||||
@ -549,17 +549,17 @@ function SourceRow({
|
||||
<span
|
||||
className={cn(
|
||||
'inline-flex items-center gap-1.5 system-xs-medium text-text-primary',
|
||||
visibleSource.status === 'syncing' && 'text-text-accent',
|
||||
source.status === 'syncing' && 'text-text-accent',
|
||||
)}
|
||||
>
|
||||
<StatusDot
|
||||
status={statusDotStatus[visibleSource.status]}
|
||||
status={statusDotStatus[source.status]}
|
||||
className={cn(
|
||||
'shrink-0',
|
||||
visibleSource.status === 'syncing' && 'animate-pulse motion-reduce:animate-none',
|
||||
source.status === 'syncing' && 'animate-pulse motion-reduce:animate-none',
|
||||
)}
|
||||
/>
|
||||
{t(($) => $[`newKnowledge.sourceStatus.${visibleSource.status}`])}
|
||||
{t(($) => $[`newKnowledge.sourceStatus.${source.status}`])}
|
||||
</span>
|
||||
</td>
|
||||
<td className="py-2 pr-3 system-xs-regular whitespace-nowrap text-text-secondary">
|
||||
@ -568,10 +568,10 @@ function SourceRow({
|
||||
<td
|
||||
className={cn(
|
||||
'py-2 pr-3 system-xs-regular whitespace-nowrap',
|
||||
visibleSource.status === 'error' ? 'text-text-destructive' : 'text-text-secondary',
|
||||
source.status === 'error' ? 'text-text-destructive' : 'text-text-secondary',
|
||||
)}
|
||||
>
|
||||
{visibleSource.status === 'syncing' && syncWorkflow ? (
|
||||
{source.status === 'syncing' && syncWorkflow ? (
|
||||
<span className="inline-flex items-center gap-1.5 text-text-accent">
|
||||
<span
|
||||
aria-hidden
|
||||
@ -585,7 +585,7 @@ function SourceRow({
|
||||
total: syncWorkflow.progressTotal ?? '—',
|
||||
})}
|
||||
</span>
|
||||
) : visibleSource.status === 'error' ? (
|
||||
) : source.status === 'error' ? (
|
||||
<span className="inline-flex items-center gap-1.5">
|
||||
<span aria-hidden className="i-ri-error-warning-fill size-3.5" />
|
||||
{syncWorkflow?.lastErrorCode ?? t(($) => $['newKnowledge.sourceSyncFailed'])}
|
||||
@ -596,7 +596,7 @@ function SourceRow({
|
||||
</td>
|
||||
<td className="py-2 text-right whitespace-nowrap">
|
||||
<div className="flex items-center justify-end gap-1">
|
||||
{canSync && visibleSource.status === 'error' && (
|
||||
{canSync && source.status === 'error' && (
|
||||
<Button
|
||||
size="small"
|
||||
variant="secondary"
|
||||
@ -610,7 +610,7 @@ function SourceRow({
|
||||
<SourceActions
|
||||
canEdit={canEdit}
|
||||
canSync={canSync}
|
||||
source={visibleSource}
|
||||
source={source}
|
||||
pendingAction={pendingAction}
|
||||
onSync={syncSource}
|
||||
onToggle={toggleSource}
|
||||
@ -732,8 +732,9 @@ export function SourcesPage({ knowledgeSpaceId }: { knowledgeSpaceId: string })
|
||||
page.data.some(
|
||||
(source) =>
|
||||
!removedSourceIds.has(source.id) &&
|
||||
getCurrentSource(sourceFromApi(source), sourceOverrides[source.id]).status ===
|
||||
'syncing',
|
||||
sourceNeedsPolling(
|
||||
getCurrentSource(sourceFromApi(source), sourceOverrides[source.id]),
|
||||
),
|
||||
),
|
||||
)
|
||||
? SOURCE_POLL_INTERVAL
|
||||
@ -990,14 +991,6 @@ export function SourcesPage({ knowledgeSpaceId }: { knowledgeSpaceId: string })
|
||||
[updatedSource.id]: updatedSource,
|
||||
}))
|
||||
}
|
||||
onSourceReconciled={() =>
|
||||
setSourceOverrides((current) => {
|
||||
if (!current[source.id]) return current
|
||||
const next = { ...current }
|
||||
delete next[source.id]
|
||||
return next
|
||||
})
|
||||
}
|
||||
onCheckedChange={(checked) => {
|
||||
setSelectedSourceIds((current) => {
|
||||
const next = new Set(current)
|
||||
|
||||
Loading…
Reference in New Issue
Block a user