slices.network
Highly ambitious ATProtocol AppView service and sdks
add batched logging for syncs, show log results per sync job in /frontend, make records actually per slice
chadtmiller.com
+1874
-221
+23
api/.sqlx/query-0747fa767581fddc88ecbb65eb975030a8426e586499f4bf9fa31370cc8c184a.json
+23
api/.sqlx/query-0747fa767581fddc88ecbb65eb975030a8426e586499f4bf9fa31370cc8c184a.json
···
1
+
{
2
+
"db_name": "PostgreSQL",
3
+
"query": "\n SELECT m.id\n FROM mq_msgs m\n JOIN mq_payloads p ON m.id = p.id\n LEFT JOIN job_results jr ON (p.payload_json->>'job_id')::uuid = jr.job_id\n WHERE m.channel_name = 'sync_queue'\n AND m.id != '00000000-0000-0000-0000-000000000000'\n AND p.payload_json->>'user_did' = $1 \n AND p.payload_json->>'slice_uri' = $2\n AND m.created_at > NOW() - INTERVAL '10 minutes'\n AND jr.job_id IS NULL\n ",
4
+
"describe": {
5
+
"columns": [
6
+
{
7
+
"ordinal": 0,
8
+
"name": "id",
9
+
"type_info": "Uuid"
10
+
}
11
+
],
12
+
"parameters": {
13
+
"Left": [
14
+
"Text",
15
+
"Text"
16
+
]
17
+
},
18
+
"nullable": [
19
+
false
20
+
]
21
+
},
22
+
"hash": "0747fa767581fddc88ecbb65eb975030a8426e586499f4bf9fa31370cc8c184a"
23
+
}
+71
api/.sqlx/query-2179459ed198e71e884c26cf1037b7916d54bf725f79e1c3f0c10ada963056c9.json
+71
api/.sqlx/query-2179459ed198e71e884c26cf1037b7916d54bf725f79e1c3f0c10ada963056c9.json
···
1
+
{
2
+
"db_name": "PostgreSQL",
3
+
"query": "\n SELECT id, created_at, log_type, job_id, user_did, slice_uri, level, message, metadata\n FROM logs\n WHERE log_type = 'sync_job' AND job_id = $1\n ORDER BY created_at ASC\n LIMIT $2\n ",
4
+
"describe": {
5
+
"columns": [
6
+
{
7
+
"ordinal": 0,
8
+
"name": "id",
9
+
"type_info": "Int8"
10
+
},
11
+
{
12
+
"ordinal": 1,
13
+
"name": "created_at",
14
+
"type_info": "Timestamptz"
15
+
},
16
+
{
17
+
"ordinal": 2,
18
+
"name": "log_type",
19
+
"type_info": "Varchar"
20
+
},
21
+
{
22
+
"ordinal": 3,
23
+
"name": "job_id",
24
+
"type_info": "Uuid"
25
+
},
26
+
{
27
+
"ordinal": 4,
28
+
"name": "user_did",
29
+
"type_info": "Text"
30
+
},
31
+
{
32
+
"ordinal": 5,
33
+
"name": "slice_uri",
34
+
"type_info": "Text"
35
+
},
36
+
{
37
+
"ordinal": 6,
38
+
"name": "level",
39
+
"type_info": "Varchar"
40
+
},
41
+
{
42
+
"ordinal": 7,
43
+
"name": "message",
44
+
"type_info": "Text"
45
+
},
46
+
{
47
+
"ordinal": 8,
48
+
"name": "metadata",
49
+
"type_info": "Jsonb"
50
+
}
51
+
],
52
+
"parameters": {
53
+
"Left": [
54
+
"Uuid",
55
+
"Int8"
56
+
]
57
+
},
58
+
"nullable": [
59
+
false,
60
+
false,
61
+
false,
62
+
true,
63
+
true,
64
+
true,
65
+
false,
66
+
false,
67
+
true
68
+
]
69
+
},
70
+
"hash": "2179459ed198e71e884c26cf1037b7916d54bf725f79e1c3f0c10ada963056c9"
71
+
}
+20
api/.sqlx/query-67967ffa1eb5f0781c2d64475bfcb5fff62cc8769feee3aa0bfe99923298b9af.json
+20
api/.sqlx/query-67967ffa1eb5f0781c2d64475bfcb5fff62cc8769feee3aa0bfe99923298b9af.json
···
1
+
{
2
+
"db_name": "PostgreSQL",
3
+
"query": "INSERT INTO \"record\" (\"uri\", \"cid\", \"did\", \"collection\", \"json\", \"indexed_at\", \"slice_uri\")\n VALUES ($1, $2, $3, $4, $5, $6, $7)\n ON CONFLICT ON CONSTRAINT record_pkey\n DO UPDATE SET\n \"cid\" = EXCLUDED.\"cid\",\n \"json\" = EXCLUDED.\"json\",\n \"indexed_at\" = EXCLUDED.\"indexed_at\"",
4
+
"describe": {
5
+
"columns": [],
6
+
"parameters": {
7
+
"Left": [
8
+
"Text",
9
+
"Text",
10
+
"Text",
11
+
"Text",
12
+
"Jsonb",
13
+
"Timestamptz",
14
+
"Text"
15
+
]
16
+
},
17
+
"nullable": []
18
+
},
19
+
"hash": "67967ffa1eb5f0781c2d64475bfcb5fff62cc8769feee3aa0bfe99923298b9af"
20
+
}
+72
api/.sqlx/query-6876227561a83034ad7c0a271b94e91972c2479f80217251ffc8ee4cc64d402e.json
+72
api/.sqlx/query-6876227561a83034ad7c0a271b94e91972c2479f80217251ffc8ee4cc64d402e.json
···
1
+
{
2
+
"db_name": "PostgreSQL",
3
+
"query": "\n SELECT id, created_at, log_type, job_id, user_did, slice_uri, level, message, metadata\n FROM logs\n WHERE slice_uri = $1 AND log_type = $2\n ORDER BY created_at DESC\n LIMIT $3\n ",
4
+
"describe": {
5
+
"columns": [
6
+
{
7
+
"ordinal": 0,
8
+
"name": "id",
9
+
"type_info": "Int8"
10
+
},
11
+
{
12
+
"ordinal": 1,
13
+
"name": "created_at",
14
+
"type_info": "Timestamptz"
15
+
},
16
+
{
17
+
"ordinal": 2,
18
+
"name": "log_type",
19
+
"type_info": "Varchar"
20
+
},
21
+
{
22
+
"ordinal": 3,
23
+
"name": "job_id",
24
+
"type_info": "Uuid"
25
+
},
26
+
{
27
+
"ordinal": 4,
28
+
"name": "user_did",
29
+
"type_info": "Text"
30
+
},
31
+
{
32
+
"ordinal": 5,
33
+
"name": "slice_uri",
34
+
"type_info": "Text"
35
+
},
36
+
{
37
+
"ordinal": 6,
38
+
"name": "level",
39
+
"type_info": "Varchar"
40
+
},
41
+
{
42
+
"ordinal": 7,
43
+
"name": "message",
44
+
"type_info": "Text"
45
+
},
46
+
{
47
+
"ordinal": 8,
48
+
"name": "metadata",
49
+
"type_info": "Jsonb"
50
+
}
51
+
],
52
+
"parameters": {
53
+
"Left": [
54
+
"Text",
55
+
"Text",
56
+
"Int8"
57
+
]
58
+
},
59
+
"nullable": [
60
+
false,
61
+
false,
62
+
false,
63
+
true,
64
+
true,
65
+
true,
66
+
false,
67
+
false,
68
+
true
69
+
]
70
+
},
71
+
"hash": "6876227561a83034ad7c0a271b94e91972c2479f80217251ffc8ee4cc64d402e"
72
+
}
+12
api/.sqlx/query-95d7cef76ba9be6cf67cd334c412e5dc9fb97b2d03d54feb1b2882aa491e4398.json
+12
api/.sqlx/query-95d7cef76ba9be6cf67cd334c412e5dc9fb97b2d03d54feb1b2882aa491e4398.json
···
1
+
{
2
+
"db_name": "PostgreSQL",
3
+
"query": "\n DELETE FROM logs\n WHERE\n (log_type = 'jetstream' AND created_at < NOW() - INTERVAL '30 days')\n OR (log_type = 'sync_job' AND created_at < NOW() - INTERVAL '7 days')\n OR (log_type = 'system' AND created_at < NOW() - INTERVAL '7 days')\n ",
4
+
"describe": {
5
+
"columns": [],
6
+
"parameters": {
7
+
"Left": []
8
+
},
9
+
"nullable": []
10
+
},
11
+
"hash": "95d7cef76ba9be6cf67cd334c412e5dc9fb97b2d03d54feb1b2882aa491e4398"
12
+
}
+3
-2
api/.sqlx/query-ac7719747ea4c2e470d97f3ff85657e8fd4ee67323ad72b606bfad3d5bc72363.json
api/.sqlx/query-573d2fa425b98e4f8ae887697725988d4cade74975d90f5e8a22c6fe7dc162a3.json
+3
-2
api/.sqlx/query-ac7719747ea4c2e470d97f3ff85657e8fd4ee67323ad72b606bfad3d5bc72363.json
api/.sqlx/query-573d2fa425b98e4f8ae887697725988d4cade74975d90f5e8a22c6fe7dc162a3.json
···
1
1
{
2
2
"db_name": "PostgreSQL",
3
-
"query": "UPDATE \"record\"\n SET \"cid\" = $1, \"json\" = $2, \"indexed_at\" = $3\n WHERE \"uri\" = $4",
3
+
"query": "UPDATE \"record\"\n SET \"cid\" = $1, \"json\" = $2, \"indexed_at\" = $3\n WHERE \"uri\" = $4 AND \"slice_uri\" = $5",
4
4
"describe": {
5
5
"columns": [],
6
6
"parameters": {
···
8
8
"Text",
9
9
"Jsonb",
10
10
"Timestamptz",
11
+
"Text",
11
12
"Text"
12
13
]
13
14
},
14
15
"nullable": []
15
16
},
16
-
"hash": "ac7719747ea4c2e470d97f3ff85657e8fd4ee67323ad72b606bfad3d5bc72363"
17
+
"hash": "573d2fa425b98e4f8ae887697725988d4cade74975d90f5e8a22c6fe7dc162a3"
17
18
}
-14
api/.sqlx/query-b64b547970d24bc3877a08c7e123647967b22f22d4e5110603a03917d1affac5.json
-14
api/.sqlx/query-b64b547970d24bc3877a08c7e123647967b22f22d4e5110603a03917d1affac5.json
···
1
-
{
2
-
"db_name": "PostgreSQL",
3
-
"query": "DELETE FROM \"record\" WHERE \"uri\" = $1",
4
-
"describe": {
5
-
"columns": [],
6
-
"parameters": {
7
-
"Left": [
8
-
"Text"
9
-
]
10
-
},
11
-
"nullable": []
12
-
},
13
-
"hash": "b64b547970d24bc3877a08c7e123647967b22f22d4e5110603a03917d1affac5"
14
-
}
+23
api/.sqlx/query-e638f20526ad4a43b0a20f0006913f767e1eacce2b26c5b692ddcdf7f9d4b629.json
+23
api/.sqlx/query-e638f20526ad4a43b0a20f0006913f767e1eacce2b26c5b692ddcdf7f9d4b629.json
···
1
+
{
2
+
"db_name": "PostgreSQL",
3
+
"query": "\n SELECT m.id \n FROM mq_msgs m\n JOIN mq_payloads p ON m.id = p.id\n WHERE m.channel_name = 'sync_queue'\n AND m.id != '00000000-0000-0000-0000-000000000000'\n AND p.payload_json->>'user_did' = $1 \n AND p.payload_json->>'slice_uri' = $2\n AND m.attempt_at <= NOW()\n ",
4
+
"describe": {
5
+
"columns": [
6
+
{
7
+
"ordinal": 0,
8
+
"name": "id",
9
+
"type_info": "Uuid"
10
+
}
11
+
],
12
+
"parameters": {
13
+
"Left": [
14
+
"Text",
15
+
"Text"
16
+
]
17
+
},
18
+
"nullable": [
19
+
false
20
+
]
21
+
},
22
+
"hash": "e638f20526ad4a43b0a20f0006913f767e1eacce2b26c5b692ddcdf7f9d4b629"
23
+
}
+70
api/.sqlx/query-eaf55dfbc38c83a834edcd368adbba7593a2e91f7a65765027c3139c42cdaec2.json
+70
api/.sqlx/query-eaf55dfbc38c83a834edcd368adbba7593a2e91f7a65765027c3139c42cdaec2.json
···
1
+
{
2
+
"db_name": "PostgreSQL",
3
+
"query": "\n SELECT id, created_at, log_type, job_id, user_did, slice_uri, level, message, metadata\n FROM logs\n WHERE log_type = 'jetstream'\n ORDER BY created_at DESC\n LIMIT $1\n ",
4
+
"describe": {
5
+
"columns": [
6
+
{
7
+
"ordinal": 0,
8
+
"name": "id",
9
+
"type_info": "Int8"
10
+
},
11
+
{
12
+
"ordinal": 1,
13
+
"name": "created_at",
14
+
"type_info": "Timestamptz"
15
+
},
16
+
{
17
+
"ordinal": 2,
18
+
"name": "log_type",
19
+
"type_info": "Varchar"
20
+
},
21
+
{
22
+
"ordinal": 3,
23
+
"name": "job_id",
24
+
"type_info": "Uuid"
25
+
},
26
+
{
27
+
"ordinal": 4,
28
+
"name": "user_did",
29
+
"type_info": "Text"
30
+
},
31
+
{
32
+
"ordinal": 5,
33
+
"name": "slice_uri",
34
+
"type_info": "Text"
35
+
},
36
+
{
37
+
"ordinal": 6,
38
+
"name": "level",
39
+
"type_info": "Varchar"
40
+
},
41
+
{
42
+
"ordinal": 7,
43
+
"name": "message",
44
+
"type_info": "Text"
45
+
},
46
+
{
47
+
"ordinal": 8,
48
+
"name": "metadata",
49
+
"type_info": "Jsonb"
50
+
}
51
+
],
52
+
"parameters": {
53
+
"Left": [
54
+
"Int8"
55
+
]
56
+
},
57
+
"nullable": [
58
+
false,
59
+
false,
60
+
false,
61
+
true,
62
+
true,
63
+
true,
64
+
false,
65
+
false,
66
+
true
67
+
]
68
+
},
69
+
"hash": "eaf55dfbc38c83a834edcd368adbba7593a2e91f7a65765027c3139c42cdaec2"
70
+
}
+71
api/.sqlx/query-fd59f79335197ad5cadd529c27ff06222e2cbd31bf88fe534863bcb5031e804b.json
+71
api/.sqlx/query-fd59f79335197ad5cadd529c27ff06222e2cbd31bf88fe534863bcb5031e804b.json
···
1
+
{
2
+
"db_name": "PostgreSQL",
3
+
"query": "\n SELECT id, created_at, log_type, job_id, user_did, slice_uri, level, message, metadata\n FROM logs\n WHERE slice_uri = $1\n ORDER BY created_at DESC\n LIMIT $2\n ",
4
+
"describe": {
5
+
"columns": [
6
+
{
7
+
"ordinal": 0,
8
+
"name": "id",
9
+
"type_info": "Int8"
10
+
},
11
+
{
12
+
"ordinal": 1,
13
+
"name": "created_at",
14
+
"type_info": "Timestamptz"
15
+
},
16
+
{
17
+
"ordinal": 2,
18
+
"name": "log_type",
19
+
"type_info": "Varchar"
20
+
},
21
+
{
22
+
"ordinal": 3,
23
+
"name": "job_id",
24
+
"type_info": "Uuid"
25
+
},
26
+
{
27
+
"ordinal": 4,
28
+
"name": "user_did",
29
+
"type_info": "Text"
30
+
},
31
+
{
32
+
"ordinal": 5,
33
+
"name": "slice_uri",
34
+
"type_info": "Text"
35
+
},
36
+
{
37
+
"ordinal": 6,
38
+
"name": "level",
39
+
"type_info": "Varchar"
40
+
},
41
+
{
42
+
"ordinal": 7,
43
+
"name": "message",
44
+
"type_info": "Text"
45
+
},
46
+
{
47
+
"ordinal": 8,
48
+
"name": "metadata",
49
+
"type_info": "Jsonb"
50
+
}
51
+
],
52
+
"parameters": {
53
+
"Left": [
54
+
"Text",
55
+
"Int8"
56
+
]
57
+
},
58
+
"nullable": [
59
+
false,
60
+
false,
61
+
false,
62
+
true,
63
+
true,
64
+
true,
65
+
false,
66
+
false,
67
+
true
68
+
]
69
+
},
70
+
"hash": "fd59f79335197ad5cadd529c27ff06222e2cbd31bf88fe534863bcb5031e804b"
71
+
}
-20
api/.sqlx/query-fe1819e244ecc16440c2914682691d5a0dbf63a4d80ce2b8799e7e2606c15ea0.json
-20
api/.sqlx/query-fe1819e244ecc16440c2914682691d5a0dbf63a4d80ce2b8799e7e2606c15ea0.json
···
1
-
{
2
-
"db_name": "PostgreSQL",
3
-
"query": "INSERT INTO \"record\" (\"uri\", \"cid\", \"did\", \"collection\", \"json\", \"indexed_at\", \"slice_uri\")\n VALUES ($1, $2, $3, $4, $5, $6, $7)\n ON CONFLICT (\"uri\")\n DO UPDATE SET\n \"cid\" = EXCLUDED.\"cid\",\n \"json\" = EXCLUDED.\"json\",\n \"indexed_at\" = EXCLUDED.\"indexed_at\",\n \"slice_uri\" = EXCLUDED.\"slice_uri\"",
4
-
"describe": {
5
-
"columns": [],
6
-
"parameters": {
7
-
"Left": [
8
-
"Text",
9
-
"Text",
10
-
"Text",
11
-
"Text",
12
-
"Jsonb",
13
-
"Timestamptz",
14
-
"Text"
15
-
]
16
-
},
17
-
"nullable": []
18
-
},
19
-
"hash": "fe1819e244ecc16440c2914682691d5a0dbf63a4d80ce2b8799e7e2606c15ea0"
20
-
}
+25
api/migrations/008_create_logs_table.sql
+25
api/migrations/008_create_logs_table.sql
···
1
+
-- Create logs table for sync jobs and jetstream activity
2
+
CREATE TABLE logs (
3
+
id BIGSERIAL PRIMARY KEY,
4
+
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
5
+
log_type VARCHAR(50) NOT NULL, -- 'sync_job', 'jetstream', etc.
6
+
job_id UUID NULL, -- For sync job logs, null for jetstream logs
7
+
user_did TEXT NULL, -- User associated with the log (for filtering)
8
+
slice_uri TEXT NULL, -- Slice associated with the log (for filtering)
9
+
level VARCHAR(20) NOT NULL DEFAULT 'info', -- 'debug', 'info', 'warn', 'error'
10
+
message TEXT NOT NULL,
11
+
metadata JSONB NULL -- Additional structured data (counts, errors, etc.)
12
+
);
13
+
14
+
-- Create indexes for efficient queries
15
+
CREATE INDEX idx_logs_type_job_id ON logs (log_type, job_id);
16
+
CREATE INDEX idx_logs_type_created_at ON logs (log_type, created_at);
17
+
CREATE INDEX idx_logs_user_did ON logs (user_did);
18
+
CREATE INDEX idx_logs_slice_uri ON logs (slice_uri);
19
+
20
+
-- Add some helpful comments
21
+
COMMENT ON TABLE logs IS 'Unified logging table for sync jobs, jetstream, and other system activities';
22
+
COMMENT ON COLUMN logs.log_type IS 'Type of log entry: sync_job, jetstream, system, etc.';
23
+
COMMENT ON COLUMN logs.job_id IS 'Associated job ID for sync job logs, null for other log types';
24
+
COMMENT ON COLUMN logs.level IS 'Log level: debug, info, warn, error';
25
+
COMMENT ON COLUMN logs.metadata IS 'Additional structured data as JSON (progress, errors, counts, etc.)';
+11
api/migrations/009_composite_primary_key_records.sql
+11
api/migrations/009_composite_primary_key_records.sql
···
1
+
-- Change record table to use composite primary key (uri, slice_uri)
2
+
-- This allows the same record to exist in multiple slices
3
+
4
+
-- First, drop the existing primary key constraint
5
+
ALTER TABLE record DROP CONSTRAINT record_pkey;
6
+
7
+
-- Add the new composite primary key
8
+
ALTER TABLE record ADD CONSTRAINT record_pkey PRIMARY KEY (uri, slice_uri);
9
+
10
+
-- Update the unique index on URI to be non-unique since URIs can now appear multiple times
11
+
-- (The existing idx_record_* indexes should still work fine for queries)
+43
api/scripts/generate_typescript.ts
+43
api/scripts/generate_typescript.ts
···
393
393
type: "JobStatus[]",
394
394
});
395
395
396
+
sourceFile.addInterface({
397
+
name: "GetJobLogsParams",
398
+
isExported: true,
399
+
properties: [
400
+
{ name: "jobId", type: "string" },
401
+
{ name: "limit", type: "number", hasQuestionToken: true },
402
+
],
403
+
});
404
+
405
+
sourceFile.addInterface({
406
+
name: "GetJobLogsResponse",
407
+
isExported: true,
408
+
properties: [
409
+
{ name: "logs", type: "LogEntry[]" },
410
+
],
411
+
});
412
+
413
+
sourceFile.addInterface({
414
+
name: "LogEntry",
415
+
isExported: true,
416
+
properties: [
417
+
{ name: "id", type: "number" },
418
+
{ name: "createdAt", type: "string" },
419
+
{ name: "logType", type: "string" },
420
+
{ name: "jobId", type: "string", hasQuestionToken: true },
421
+
{ name: "userDid", type: "string", hasQuestionToken: true },
422
+
{ name: "sliceUri", type: "string", hasQuestionToken: true },
423
+
{ name: "level", type: "string" },
424
+
{ name: "message", type: "string" },
425
+
{ name: "metadata", type: "Record<string, unknown>", hasQuestionToken: true },
426
+
],
427
+
});
428
+
396
429
// Sync user collections interfaces
397
430
sourceFile.addInterface({
398
431
name: "SyncUserCollectionsRequest",
···
1607
1640
isAsync: true,
1608
1641
statements: [
1609
1642
`return await this.makeRequest<GetJobHistoryResponse>('social.slices.slice.getJobHistory', 'GET', params);`,
1643
+
],
1644
+
});
1645
+
1646
+
classDeclaration.addMethod({
1647
+
name: "getJobLogs",
1648
+
parameters: [{ name: "params", type: "GetJobLogsParams" }],
1649
+
returnType: "Promise<GetJobLogsResponse>",
1650
+
isAsync: true,
1651
+
statements: [
1652
+
`return await this.makeRequest<GetJobLogsResponse>('social.slices.slice.getJobLogs', 'GET', params);`,
1610
1653
],
1611
1654
});
1612
1655
+23
-30
api/src/database.rs
+23
-30
api/src/database.rs
···
269
269
sqlx::query!(
270
270
r#"INSERT INTO "record" ("uri", "cid", "did", "collection", "json", "indexed_at", "slice_uri")
271
271
VALUES ($1, $2, $3, $4, $5, $6, $7)
272
-
ON CONFLICT ("uri")
272
+
ON CONFLICT ON CONSTRAINT record_pkey
273
273
DO UPDATE SET
274
274
"cid" = EXCLUDED."cid",
275
275
"json" = EXCLUDED."json",
276
-
"indexed_at" = EXCLUDED."indexed_at",
277
-
"slice_uri" = EXCLUDED."slice_uri""#,
276
+
"indexed_at" = EXCLUDED."indexed_at""#,
278
277
record.uri,
279
278
record.cid,
280
279
record.did,
···
326
325
}
327
326
328
327
query.push_str(r#"
329
-
ON CONFLICT ("uri")
328
+
ON CONFLICT ON CONSTRAINT record_pkey
330
329
DO UPDATE SET
331
330
"cid" = EXCLUDED."cid",
332
331
"json" = EXCLUDED."json",
333
-
"indexed_at" = EXCLUDED."indexed_at",
334
-
"slice_uri" = EXCLUDED."slice_uri"
332
+
"indexed_at" = EXCLUDED."indexed_at"
335
333
"#);
336
334
337
335
// Bind all parameters
···
431
429
let result = sqlx::query!(
432
430
r#"UPDATE "record"
433
431
SET "cid" = $1, "json" = $2, "indexed_at" = $3
434
-
WHERE "uri" = $4"#,
432
+
WHERE "uri" = $4 AND "slice_uri" = $5"#,
435
433
record.cid,
436
434
record.json,
437
435
record.indexed_at,
438
-
record.uri
436
+
record.uri,
437
+
record.slice_uri
439
438
)
440
439
.execute(&self.pool)
441
440
.await?;
···
447
446
Ok(())
448
447
}
449
448
450
-
pub async fn delete_record(&self, uri: &str) -> Result<(), DatabaseError> {
451
-
let result = sqlx::query!(
452
-
r#"DELETE FROM "record" WHERE "uri" = $1"#,
453
-
uri
454
-
)
455
-
.execute(&self.pool)
456
-
.await?;
457
-
458
-
if result.rows_affected() == 0 {
459
-
return Err(DatabaseError::RecordNotFound { uri: uri.to_string() });
460
-
}
461
-
462
-
Ok(())
463
-
}
464
449
465
450
pub async fn batch_insert_actors(&self, actors: &[Actor]) -> Result<(), DatabaseError> {
466
451
let mut tx = self.pool.begin().await?;
···
996
981
Ok(count)
997
982
}
998
983
999
-
pub async fn delete_record_by_uri(&self, uri: &str) -> Result<u64, DatabaseError> {
1000
-
let result = sqlx::query("DELETE FROM record WHERE uri = $1")
1001
-
.bind(uri)
1002
-
.execute(&self.pool)
1003
-
.await?;
984
+
pub async fn delete_record_by_uri(&self, uri: &str, slice_uri: Option<&str>) -> Result<u64, DatabaseError> {
985
+
let result = if let Some(slice_uri) = slice_uri {
986
+
sqlx::query("DELETE FROM record WHERE uri = $1 AND slice_uri = $2")
987
+
.bind(uri)
988
+
.bind(slice_uri)
989
+
.execute(&self.pool)
990
+
.await?
991
+
} else {
992
+
// Delete from all slices if no specific slice provided
993
+
sqlx::query("DELETE FROM record WHERE uri = $1")
994
+
.bind(uri)
995
+
.execute(&self.pool)
996
+
.await?
997
+
};
1004
998
Ok(result.rows_affected())
1005
999
}
1006
1000
···
1009
1003
sqlx::query(r#"
1010
1004
INSERT INTO record (uri, cid, did, collection, json, indexed_at, slice_uri)
1011
1005
VALUES ($1, $2, $3, $4, $5, $6, $7)
1012
-
ON CONFLICT (uri) DO UPDATE
1006
+
ON CONFLICT ON CONSTRAINT record_pkey DO UPDATE
1013
1007
SET cid = EXCLUDED.cid,
1014
1008
json = EXCLUDED.json,
1015
-
indexed_at = EXCLUDED.indexed_at,
1016
-
slice_uri = EXCLUDED.slice_uri
1009
+
indexed_at = EXCLUDED.indexed_at
1017
1010
"#)
1018
1011
.bind(&record.uri)
1019
1012
.bind(&record.cid)
+54
api/src/handler_logs.rs
+54
api/src/handler_logs.rs
···
1
+
use axum::{
2
+
extract::{Query, State},
3
+
http::StatusCode,
4
+
response::Json,
5
+
};
6
+
use serde::{Deserialize, Serialize};
7
+
use uuid::Uuid;
8
+
9
+
use crate::{AppState, logging::{get_sync_job_logs, get_jetstream_logs, LogEntry}};
10
+
11
+
#[derive(Debug, Deserialize)]
12
+
pub struct LogsQuery {
13
+
pub limit: Option<i64>,
14
+
}
15
+
16
+
#[derive(Debug, Serialize)]
17
+
pub struct LogsResponse {
18
+
pub logs: Vec<LogEntry>,
19
+
}
20
+
21
+
#[derive(Debug, Deserialize)]
22
+
#[serde(rename_all = "camelCase")]
23
+
pub struct LogsQueryWithJobId {
24
+
pub job_id: Uuid,
25
+
pub limit: Option<i64>,
26
+
}
27
+
28
+
/// Get logs for a specific sync job
29
+
pub async fn get_sync_job_logs_handler(
30
+
State(state): State<AppState>,
31
+
Query(params): Query<LogsQueryWithJobId>,
32
+
) -> Result<Json<LogsResponse>, StatusCode> {
33
+
match get_sync_job_logs(&state.database_pool, params.job_id, params.limit).await {
34
+
Ok(logs) => Ok(Json(LogsResponse { logs })),
35
+
Err(e) => {
36
+
tracing::error!("Failed to get sync job logs: {}", e);
37
+
Err(StatusCode::INTERNAL_SERVER_ERROR)
38
+
}
39
+
}
40
+
}
41
+
42
+
/// Get jetstream logs
43
+
pub async fn get_jetstream_logs_handler(
44
+
State(state): State<AppState>,
45
+
Query(params): Query<LogsQuery>,
46
+
) -> Result<Json<LogsResponse>, StatusCode> {
47
+
match get_jetstream_logs(&state.database_pool, params.limit).await {
48
+
Ok(logs) => Ok(Json(LogsResponse { logs })),
49
+
Err(e) => {
50
+
tracing::error!("Failed to get jetstream logs: {}", e);
51
+
Err(StatusCode::INTERNAL_SERVER_ERROR)
52
+
}
53
+
}
54
+
}
+44
-48
api/src/handler_xrpc_dynamic.rs
+44
-48
api/src/handler_xrpc_dynamic.rs
···
23
23
StatusCode::INTERNAL_SERVER_ERROR => "Internal server error",
24
24
_ => "Request failed",
25
25
};
26
-
26
+
27
27
(status, Json(serde_json::json!({
28
28
"error": status.as_str(),
29
29
"message": message
···
37
37
pub uri: String,
38
38
pub slice: String,
39
39
}
40
-
41
-
42
-
43
-
44
40
45
41
// Dynamic XRPC handler that routes based on method name (for GET requests)
46
42
pub async fn dynamic_xrpc_handler(
···
103
99
.and_then(|v| v.as_str())
104
100
.ok_or(StatusCode::BAD_REQUEST)?
105
101
.to_string();
106
-
102
+
107
103
let limit = params.get("limit").and_then(|v| {
108
104
if let Some(s) = v.as_str() {
109
105
s.parse::<i32>().ok()
···
111
107
v.as_i64().map(|i| i as i32)
112
108
}
113
109
});
114
-
110
+
115
111
let cursor = params.get("cursor")
116
112
.and_then(|v| v.as_str())
117
113
.map(|s| s.to_string());
118
-
114
+
119
115
// Parse sortBy from params - convert legacy sort string to new array format if present
120
116
let sort_by = params.get("sort")
121
117
.and_then(|v| v.as_str())
···
139
135
}
140
136
sort_fields
141
137
});
142
-
138
+
143
139
// Parse where conditions from query params if present
144
140
let mut where_conditions = HashMap::new();
145
-
141
+
146
142
// Handle legacy author/authors params by converting to where clause
147
143
if let Some(author_str) = params.get("author").and_then(|v| v.as_str()) {
148
144
where_conditions.insert("did".to_string(), WhereCondition {
···
161
157
contains: None,
162
158
});
163
159
}
164
-
160
+
165
161
// Handle legacy query param by converting to where clause with contains
166
162
if let Some(query_str) = params.get("query").and_then(|v| v.as_str()) {
167
163
let field = params.get("field")
···
173
169
contains: Some(query_str.to_string()),
174
170
});
175
171
}
176
-
172
+
177
173
let where_clause = if where_conditions.is_empty() {
178
174
None
179
175
} else {
···
182
178
or_conditions: None,
183
179
})
184
180
};
185
-
181
+
186
182
let records_params = SliceRecordsParams {
187
183
slice,
188
184
limit,
···
190
186
where_clause,
191
187
sort_by: sort_by.clone(),
192
188
};
193
-
189
+
194
190
// First verify the collection belongs to this slice
195
191
let slice_collections = state.database.get_slice_collections_list(&records_params.slice).await
196
192
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
197
-
193
+
198
194
// Special handling: social.slices.lexicon is always allowed as it defines the schema
199
195
if collection != "social.slices.lexicon" && !slice_collections.contains(&collection) {
200
196
return Err(StatusCode::NOT_FOUND);
201
197
}
202
-
198
+
203
199
// Use the unified database method
204
200
match state.database.get_slice_collections_records(
205
201
&records_params.slice,
···
211
207
Ok((mut records, cursor)) => {
212
208
// Filter records to only include the specific collection
213
209
records.retain(|record| record.collection == collection);
214
-
210
+
215
211
let indexed_records: Vec<IndexedRecord> = records.into_iter().map(|record| IndexedRecord {
216
212
uri: record.uri,
217
213
cid: record.cid,
···
220
216
value: record.json,
221
217
indexed_at: record.indexed_at.to_rfc3339(),
222
218
}).collect();
223
-
219
+
224
220
let output = SliceRecordsOutput {
225
221
success: true,
226
222
records: indexed_records,
227
223
cursor,
228
224
message: None,
229
225
};
230
-
226
+
231
227
Ok(Json(serde_json::to_value(output)
232
228
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?))
233
229
},
···
249
245
// First verify the collection belongs to this slice
250
246
let slice_collections = state.database.get_slice_collections_list(&get_params.slice).await
251
247
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
252
-
248
+
253
249
// Special handling: social.slices.lexicon is always allowed as it defines the schema
254
250
if collection != "social.slices.lexicon" && !slice_collections.contains(&collection) {
255
251
return Err(StatusCode::NOT_FOUND);
···
284
280
// First verify the collection belongs to this slice
285
281
let slice_collections = state.database.get_slice_collections_list(&records_params.slice).await
286
282
.map_err(|_| (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({"error": "Database error"}))))?;
287
-
283
+
288
284
// Special handling: social.slices.lexicon is always allowed as it defines the schema
289
285
if collection != "social.slices.lexicon" && !slice_collections.contains(&collection) {
290
286
return Err((StatusCode::NOT_FOUND, Json(serde_json::json!({"error": "Collection not found"}))));
291
287
}
292
-
288
+
293
289
// Add collection filter to where conditions
294
290
let mut where_clause = records_params.where_clause.unwrap_or(crate::models::WhereClause {
295
291
conditions: HashMap::new(),
···
301
297
contains: None,
302
298
});
303
299
records_params.where_clause = Some(where_clause);
304
-
300
+
305
301
// Use the unified database method
306
302
match state.database.get_slice_collections_records(
307
303
&records_params.slice,
···
312
308
).await {
313
309
Ok((records, cursor)) => {
314
310
// No need to filter - collection filter is in the SQL query now
315
-
311
+
316
312
// Transform Record to IndexedRecord for the response
317
313
let indexed_records: Vec<IndexedRecord> = records.into_iter().map(|record| IndexedRecord {
318
314
uri: record.uri,
···
329
325
cursor,
330
326
message: None,
331
327
};
332
-
328
+
333
329
Ok(Json(serde_json::to_value(output)
334
330
.map_err(|_| (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({"error": "Serialization error"}))))?))
335
331
},
···
350
346
// First verify the collection belongs to this slice
351
347
let slice_collections = state.database.get_slice_collections_list(&records_params.slice).await
352
348
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
353
-
349
+
354
350
// Special handling: social.slices.lexicon is always allowed as it defines the schema
355
351
if collection != "social.slices.lexicon" && !slice_collections.contains(&collection) {
356
352
return Err(StatusCode::NOT_FOUND);
357
353
}
358
-
354
+
359
355
// Add collection filter to where conditions
360
356
let mut where_clause = records_params.where_clause.unwrap_or(crate::models::WhereClause {
361
357
conditions: HashMap::new(),
···
395
391
// First verify the collection belongs to this slice
396
392
let slice_collections = state.database.get_slice_collections_list(&records_params.slice).await
397
393
.map_err(|_| (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({"error": "Database error"}))))?;
398
-
394
+
399
395
// Special handling: social.slices.lexicon is always allowed as it defines the schema
400
396
if collection != "social.slices.lexicon" && !slice_collections.contains(&collection) {
401
397
return Err((StatusCode::NOT_FOUND, Json(serde_json::json!({"error": "Collection not found"}))));
402
398
}
403
-
399
+
404
400
// Add collection filter to where conditions
405
401
let mut where_clause = records_params.where_clause.unwrap_or(crate::models::WhereClause {
406
402
conditions: HashMap::new(),
···
454
450
.and_then(|v| v.as_str())
455
451
.ok_or_else(|| status_to_error_response(StatusCode::BAD_REQUEST))?
456
452
.to_string();
457
-
453
+
458
454
let record_key = body.get("rkey")
459
455
.and_then(|v| v.as_str())
460
456
.filter(|s| !s.is_empty()) // Filter out empty strings
461
457
.map(|s| s.to_string());
462
-
458
+
463
459
let record_data = body.get("record")
464
460
.ok_or_else(|| status_to_error_response(StatusCode::BAD_REQUEST))?
465
461
.clone();
466
-
467
-
462
+
463
+
468
464
// Validate the record against its lexicon
469
-
465
+
470
466
// For social.slices.lexicon collection, validate against the system slice
471
467
let validation_slice_uri = if collection == "social.slices.lexicon" {
472
468
"at://did:plc:bcgltzqazw5tb6k2g3ttenbj/social.slices.slice/3lwzmbjpqxk2q"
473
469
} else {
474
470
&slice_uri
475
471
};
476
-
477
-
472
+
473
+
478
474
match LexiconValidator::for_slice(&state.database, validation_slice_uri).await {
479
475
Ok(validator) => {
480
-
476
+
481
477
// Debug: Get lexicons from the system slice to see what's there
482
478
if collection == "social.slices.lexicon" {
483
479
}
484
-
480
+
485
481
if let Err(e) = validator.validate_record(&collection, &record_data) {
486
482
return Err((StatusCode::BAD_REQUEST, Json(serde_json::json!({
487
483
"error": "ValidationError",
···
496
492
}
497
493
498
494
// Create record using AT Protocol functions with DPoP
499
-
495
+
500
496
let create_request = CreateRecordRequest {
501
497
repo: repo.clone(),
502
498
collection: collection.clone(),
···
566
562
.and_then(|v| v.as_str())
567
563
.ok_or_else(|| status_to_error_response(StatusCode::BAD_REQUEST))?
568
564
.to_string();
569
-
565
+
570
566
let rkey = body.get("rkey")
571
567
.and_then(|v| v.as_str())
572
568
.ok_or_else(|| status_to_error_response(StatusCode::BAD_REQUEST))?
573
569
.to_string();
574
-
570
+
575
571
let record_data = body.get("record")
576
572
.ok_or_else(|| status_to_error_response(StatusCode::BAD_REQUEST))?
577
573
.clone();
578
574
579
575
// Extract repo from user info
580
576
let repo = user_info.did.unwrap_or(user_info.sub);
581
-
577
+
582
578
// Validate the record against its lexicon
583
579
match LexiconValidator::for_slice(&state.database, &slice_uri).await {
584
580
Ok(validator) => {
···
677
673
.await
678
674
.map_err(|_| status_to_error_response(StatusCode::INTERNAL_SERVER_ERROR))?;
679
675
680
-
// Also delete from local database
676
+
// Also delete from local database (from all slices)
681
677
let uri = format!("at://{}/{}/{}", repo, collection, rkey);
682
-
let _ = state.database.delete_record(&uri).await;
678
+
let _ = state.database.delete_record_by_uri(&uri, None).await;
683
679
684
680
Ok(Json(serde_json::json!({})))
685
681
}
···
782
778
783
779
let result = validator.validate_record("social.slices.testRecord", &invalid_record);
784
780
assert!(result.is_err(), "Invalid record should fail validation");
785
-
781
+
786
782
if let Err(e) = result {
787
783
let error_message = format!("{}", e);
788
784
assert!(!error_message.is_empty(), "Error message should not be empty");
789
785
// Error message should be user-friendly and descriptive
790
-
assert!(error_message.contains("aspectRatio") || error_message.contains("required"),
786
+
assert!(error_message.contains("aspectRatio") || error_message.contains("required"),
791
787
"Error message should indicate what's wrong: {}", error_message);
792
788
}
793
789
}
···
808
804
809
805
let result = validator.validate_record("social.slices.testRecord", &invalid_record);
810
806
assert!(result.is_err(), "Constraint violation should fail validation");
811
-
807
+
812
808
if let Err(e) = result {
813
809
let error_message = format!("{}", e);
814
810
// Should indicate the specific constraint that was violated
815
-
assert!(error_message.contains("length") || error_message.contains("maximum") || error_message.contains("100"),
811
+
assert!(error_message.contains("length") || error_message.contains("maximum") || error_message.contains("100"),
816
812
"Error message should indicate length constraint: {}", error_message);
817
813
}
818
814
}
+1
-1
api/src/jetstream.rs
+1
-1
api/src/jetstream.rs
···
264
264
// DID is an actor in our system, delete the record globally
265
265
let uri = format!("at://{}/{}/{}", did, commit.collection, commit.rkey);
266
266
267
-
match self.database.delete_record_by_uri(&uri).await {
267
+
match self.database.delete_record_by_uri(&uri, None).await {
268
268
Ok(rows_affected) => {
269
269
if rows_affected > 0 {
270
270
info!("✓ Deleted record globally: {} ({} rows)", uri, rows_affected);
+108
-5
api/src/jobs.rs
+108
-5
api/src/jobs.rs
···
4
4
use uuid::Uuid;
5
5
use crate::sync::SyncService;
6
6
use crate::models::BulkSyncParams;
7
+
use crate::logging::LogLevel;
8
+
use serde_json::json;
7
9
use tracing::{info, error};
8
10
9
11
/// Payload for sync jobs
···
26
28
pub message: String,
27
29
}
28
30
29
-
/// Initialize the job registry with all job handlers
31
+
/// Initialize the job registry with all job handlers
30
32
pub fn registry() -> JobRegistry {
31
33
JobRegistry::new(&[sync_job])
32
34
}
···
41
43
payload.job_id, payload.user_did, payload.slice_uri
42
44
);
43
45
44
-
// Get database pool from job context
46
+
// Get database pool and logger
45
47
let pool = current_job.pool();
48
+
let logger = crate::logging::Logger::global();
46
49
47
-
// Create sync service
50
+
// Log job start
51
+
logger.log_sync_job(
52
+
payload.job_id,
53
+
&payload.user_did,
54
+
&payload.slice_uri,
55
+
LogLevel::Info,
56
+
&format!("Starting sync job for {} collections",
57
+
payload.params.collections.as_ref().map(|c| c.len()).unwrap_or(0) +
58
+
payload.params.external_collections.as_ref().map(|c| c.len()).unwrap_or(0)
59
+
),
60
+
Some(json!({
61
+
"collections": payload.params.collections,
62
+
"external_collections": payload.params.external_collections,
63
+
"repos": payload.params.repos,
64
+
"skip_validation": payload.params.skip_validation
65
+
}))
66
+
);
67
+
68
+
// Create sync service with logging
48
69
let database = crate::database::Database::from_pool(pool.clone());
49
70
let relay_endpoint = std::env::var("RELAY_ENDPOINT")
50
71
.unwrap_or_else(|_| "https://relay1.us-west.bsky.network".to_string());
51
-
let sync_service = SyncService::new(database.clone(), relay_endpoint);
72
+
let sync_service = SyncService::with_logging(
73
+
database.clone(),
74
+
relay_endpoint,
75
+
logger.clone(),
76
+
payload.job_id,
77
+
payload.user_did.clone()
78
+
);
52
79
53
80
// Track progress
54
81
let start_time = std::time::Instant::now();
···
65
92
.await
66
93
{
67
94
Ok((repos_processed, records_synced)) => {
95
+
let elapsed = start_time.elapsed();
68
96
let result = SyncJobResult {
69
97
success: true,
70
98
total_records: records_synced,
···
75
103
repos_processed,
76
104
message: format!(
77
105
"Sync completed successfully in {:?}",
78
-
start_time.elapsed()
106
+
elapsed
79
107
),
80
108
};
81
109
110
+
// Log successful completion
111
+
logger.log_sync_job(
112
+
payload.job_id,
113
+
&payload.user_did,
114
+
&payload.slice_uri,
115
+
LogLevel::Info,
116
+
&format!("Sync completed successfully: {} repos, {} records in {:?}",
117
+
repos_processed, records_synced, elapsed),
118
+
Some(json!({
119
+
"repos_processed": repos_processed,
120
+
"records_synced": records_synced,
121
+
"duration_secs": elapsed.as_secs_f64(),
122
+
"collections_synced": result.collections_synced
123
+
}))
124
+
);
125
+
82
126
// Store result in database before completing the job
83
127
store_job_result(
84
128
pool,
···
107
151
Err(e) => {
108
152
error!("Sync job {} failed: {}", payload.job_id, e);
109
153
154
+
// Log error
155
+
logger.log_sync_job(
156
+
payload.job_id,
157
+
&payload.user_did,
158
+
&payload.slice_uri,
159
+
LogLevel::Error,
160
+
&format!("Sync job failed: {}", e),
161
+
Some(json!({
162
+
"error": e.to_string(),
163
+
"duration_secs": start_time.elapsed().as_secs_f64()
164
+
}))
165
+
);
166
+
110
167
let result = SyncJobResult {
111
168
success: false,
112
169
total_records: 0,
···
191
248
slice_uri: String,
192
249
params: BulkSyncParams,
193
250
) -> Result<Uuid, Box<dyn std::error::Error + Send + Sync>> {
251
+
// Check if there's already a running sync job for this user+slice combination
252
+
// We do this by checking:
253
+
// 1. If there are any jobs in mq_msgs for sync_queue channel that haven't been processed yet
254
+
// 2. If there are any recent job_results entries that indicate a job might still be running
255
+
let existing_running_msg = sqlx::query!(
256
+
r#"
257
+
SELECT m.id
258
+
FROM mq_msgs m
259
+
JOIN mq_payloads p ON m.id = p.id
260
+
WHERE m.channel_name = 'sync_queue'
261
+
AND m.id != '00000000-0000-0000-0000-000000000000'
262
+
AND p.payload_json->>'user_did' = $1
263
+
AND p.payload_json->>'slice_uri' = $2
264
+
AND m.attempt_at <= NOW()
265
+
"#,
266
+
user_did,
267
+
slice_uri
268
+
)
269
+
.fetch_optional(pool)
270
+
.await?;
271
+
272
+
// Also check if there's a very recent job that might still be running
273
+
// (within the last 10 minutes and no completion record)
274
+
let recent_start = sqlx::query!(
275
+
r#"
276
+
SELECT m.id
277
+
FROM mq_msgs m
278
+
JOIN mq_payloads p ON m.id = p.id
279
+
LEFT JOIN job_results jr ON (p.payload_json->>'job_id')::uuid = jr.job_id
280
+
WHERE m.channel_name = 'sync_queue'
281
+
AND m.id != '00000000-0000-0000-0000-000000000000'
282
+
AND p.payload_json->>'user_did' = $1
283
+
AND p.payload_json->>'slice_uri' = $2
284
+
AND m.created_at > NOW() - INTERVAL '10 minutes'
285
+
AND jr.job_id IS NULL
286
+
"#,
287
+
user_did,
288
+
slice_uri
289
+
)
290
+
.fetch_optional(pool)
291
+
.await?;
292
+
293
+
if existing_running_msg.is_some() || recent_start.is_some() {
294
+
return Err("A sync job is already running for this slice. Please wait for it to complete before starting another.".into());
295
+
}
296
+
194
297
let job_id = Uuid::new_v4();
195
298
196
299
let payload = SyncJobPayload {
+353
api/src/logging.rs
+353
api/src/logging.rs
···
1
+
use serde_json::Value;
2
+
use sqlx::PgPool;
3
+
use uuid::Uuid;
4
+
use tokio::sync::mpsc;
5
+
use tokio::time::{interval, Duration};
6
+
use tracing::{info, warn, error};
7
+
use chrono::Utc;
8
+
use std::sync::OnceLock;
9
+
10
+
#[derive(Debug, Clone)]
11
+
pub enum LogLevel {
12
+
Info,
13
+
Warn,
14
+
Error,
15
+
}
16
+
17
+
impl LogLevel {
18
+
pub fn as_str(&self) -> &'static str {
19
+
match self {
20
+
LogLevel::Info => "info",
21
+
LogLevel::Warn => "warn",
22
+
LogLevel::Error => "error",
23
+
}
24
+
}
25
+
}
26
+
27
+
#[derive(Debug, Clone)]
28
+
#[allow(dead_code)]
29
+
pub enum LogType {
30
+
SyncJob,
31
+
Jetstream,
32
+
System,
33
+
}
34
+
35
+
impl LogType {
36
+
pub fn as_str(&self) -> &'static str {
37
+
match self {
38
+
LogType::SyncJob => "sync_job",
39
+
LogType::Jetstream => "jetstream",
40
+
LogType::System => "system",
41
+
}
42
+
}
43
+
}
44
+
45
+
/// Global logger instance
46
+
static GLOBAL_LOGGER: OnceLock<Logger> = OnceLock::new();
47
+
48
+
/// Log entry to be queued for batch insertion
49
+
#[derive(Debug, Clone)]
50
+
struct QueuedLogEntry {
51
+
log_type: String,
52
+
job_id: Option<Uuid>,
53
+
user_did: Option<String>,
54
+
slice_uri: Option<String>,
55
+
level: String,
56
+
message: String,
57
+
metadata: Option<Value>,
58
+
created_at: chrono::DateTime<chrono::Utc>,
59
+
}
60
+
61
+
/// Logger that queues log entries and flushes them periodically
62
+
#[derive(Clone)]
63
+
pub struct Logger {
64
+
sender: mpsc::UnboundedSender<QueuedLogEntry>,
65
+
}
66
+
67
+
impl Logger {
68
+
/// Create a new batched logger and spawn the background worker
69
+
pub fn new(pool: PgPool) -> Self {
70
+
let (sender, receiver) = mpsc::unbounded_channel();
71
+
72
+
// Spawn background worker
73
+
tokio::spawn(Self::background_worker(receiver, pool));
74
+
75
+
Self { sender }
76
+
}
77
+
78
+
/// Initialize the global logger (call once at startup)
79
+
pub fn init_global(pool: PgPool) {
80
+
let logger = Self::new(pool);
81
+
if GLOBAL_LOGGER.set(logger).is_err() {
82
+
warn!("Global logger was already initialized");
83
+
}
84
+
}
85
+
86
+
/// Get the global logger instance
87
+
pub fn global() -> &'static Logger {
88
+
GLOBAL_LOGGER.get().expect("Global logger not initialized - call Logger::init_global() first")
89
+
}
90
+
91
+
/// Log a sync job message (queued for batch insertion)
92
+
pub fn log_sync_job(
93
+
&self,
94
+
job_id: Uuid,
95
+
user_did: &str,
96
+
slice_uri: &str,
97
+
level: LogLevel,
98
+
message: &str,
99
+
metadata: Option<Value>,
100
+
) {
101
+
let entry = QueuedLogEntry {
102
+
log_type: LogType::SyncJob.as_str().to_string(),
103
+
job_id: Some(job_id),
104
+
user_did: Some(user_did.to_string()),
105
+
slice_uri: Some(slice_uri.to_string()),
106
+
level: level.as_str().to_string(),
107
+
message: message.to_string(),
108
+
metadata,
109
+
created_at: Utc::now(),
110
+
};
111
+
112
+
// Also log to tracing for immediate console output
113
+
match level {
114
+
LogLevel::Info => info!("[sync_job] {}", message),
115
+
LogLevel::Warn => warn!("[sync_job] {}", message),
116
+
LogLevel::Error => error!("[sync_job] {}", message),
117
+
}
118
+
119
+
// Queue for database insertion (ignore send errors if channel closed)
120
+
let _ = self.sender.send(entry);
121
+
}
122
+
123
+
/// Background worker that processes the log queue
124
+
async fn background_worker(
125
+
mut receiver: mpsc::UnboundedReceiver<QueuedLogEntry>,
126
+
pool: PgPool,
127
+
) {
128
+
let mut batch = Vec::new();
129
+
let mut flush_interval = interval(Duration::from_secs(5)); // Flush every 5 seconds
130
+
131
+
info!("Started batched logging background worker");
132
+
133
+
loop {
134
+
tokio::select! {
135
+
// Receive log entries
136
+
Some(entry) = receiver.recv() => {
137
+
batch.push(entry);
138
+
139
+
// Flush if batch is large enough
140
+
if batch.len() >= 100 {
141
+
Self::flush_batch(&pool, &mut batch).await;
142
+
}
143
+
}
144
+
145
+
// Periodic flush
146
+
_ = flush_interval.tick() => {
147
+
if !batch.is_empty() {
148
+
Self::flush_batch(&pool, &mut batch).await;
149
+
}
150
+
}
151
+
152
+
// Channel closed, flush remaining and exit
153
+
else => {
154
+
if !batch.is_empty() {
155
+
Self::flush_batch(&pool, &mut batch).await;
156
+
}
157
+
break;
158
+
}
159
+
}
160
+
}
161
+
162
+
info!("Batched logging background worker shut down");
163
+
}
164
+
165
+
/// Flush a batch of log entries to the database
166
+
async fn flush_batch(pool: &PgPool, batch: &mut Vec<QueuedLogEntry>) {
167
+
if batch.is_empty() {
168
+
return;
169
+
}
170
+
171
+
let batch_size = batch.len();
172
+
let start = std::time::Instant::now();
173
+
174
+
// Build bulk INSERT query
175
+
let mut query = String::from(
176
+
"INSERT INTO logs (log_type, job_id, user_did, slice_uri, level, message, metadata, created_at) VALUES "
177
+
);
178
+
179
+
// Add placeholders for each record
180
+
for i in 0..batch_size {
181
+
if i > 0 {
182
+
query.push_str(", ");
183
+
}
184
+
let base = i * 8 + 1; // 8 fields per log entry
185
+
query.push_str(&format!(
186
+
"(${}, ${}, ${}, ${}, ${}, ${}, ${}, ${})",
187
+
base, base + 1, base + 2, base + 3, base + 4, base + 5, base + 6, base + 7
188
+
));
189
+
}
190
+
191
+
// Bind parameters
192
+
let mut sqlx_query = sqlx::query(&query);
193
+
for entry in batch.iter() {
194
+
sqlx_query = sqlx_query
195
+
.bind(&entry.log_type)
196
+
.bind(&entry.job_id)
197
+
.bind(&entry.user_did)
198
+
.bind(&entry.slice_uri)
199
+
.bind(&entry.level)
200
+
.bind(&entry.message)
201
+
.bind(&entry.metadata)
202
+
.bind(&entry.created_at);
203
+
}
204
+
205
+
// Execute batch insert
206
+
match sqlx_query.execute(pool).await {
207
+
Ok(_) => {
208
+
let elapsed = start.elapsed();
209
+
if elapsed.as_millis() > 100 {
210
+
warn!("Slow log batch insert: {} entries in {:?}", batch_size, elapsed);
211
+
} else {
212
+
info!("Flushed {} log entries in {:?}", batch_size, elapsed);
213
+
}
214
+
}
215
+
Err(e) => {
216
+
error!("Failed to flush log batch of {} entries: {}", batch_size, e);
217
+
// Continue processing - logs are lost but system keeps running
218
+
}
219
+
}
220
+
221
+
batch.clear();
222
+
}
223
+
}
224
+
225
+
/// Log entry struct for database queries
226
+
#[derive(Debug, serde::Serialize, sqlx::FromRow)]
227
+
#[serde(rename_all = "camelCase")]
228
+
pub struct LogEntry {
229
+
pub id: i64,
230
+
pub created_at: chrono::DateTime<chrono::Utc>,
231
+
pub log_type: String,
232
+
pub job_id: Option<Uuid>,
233
+
pub user_did: Option<String>,
234
+
pub slice_uri: Option<String>,
235
+
pub level: String,
236
+
pub message: String,
237
+
pub metadata: Option<serde_json::Value>,
238
+
}
239
+
240
+
/// Get logs for a specific sync job
241
+
pub async fn get_sync_job_logs(
242
+
pool: &PgPool,
243
+
job_id: Uuid,
244
+
limit: Option<i64>,
245
+
) -> Result<Vec<LogEntry>, sqlx::Error> {
246
+
let limit = limit.unwrap_or(100);
247
+
248
+
let rows = sqlx::query_as!(
249
+
LogEntry,
250
+
r#"
251
+
SELECT id, created_at, log_type, job_id, user_did, slice_uri, level, message, metadata
252
+
FROM logs
253
+
WHERE log_type = 'sync_job' AND job_id = $1
254
+
ORDER BY created_at ASC
255
+
LIMIT $2
256
+
"#,
257
+
job_id,
258
+
limit
259
+
)
260
+
.fetch_all(pool)
261
+
.await?;
262
+
263
+
Ok(rows)
264
+
}
265
+
266
+
/// Get jetstream logs
267
+
#[allow(dead_code)]
268
+
pub async fn get_jetstream_logs(
269
+
pool: &PgPool,
270
+
limit: Option<i64>,
271
+
) -> Result<Vec<LogEntry>, sqlx::Error> {
272
+
let limit = limit.unwrap_or(100);
273
+
274
+
let rows = sqlx::query_as!(
275
+
LogEntry,
276
+
r#"
277
+
SELECT id, created_at, log_type, job_id, user_did, slice_uri, level, message, metadata
278
+
FROM logs
279
+
WHERE log_type = 'jetstream'
280
+
ORDER BY created_at DESC
281
+
LIMIT $1
282
+
"#,
283
+
limit
284
+
)
285
+
.fetch_all(pool)
286
+
.await?;
287
+
288
+
Ok(rows)
289
+
}
290
+
291
+
/// Get logs for a specific slice
292
+
#[allow(dead_code)]
293
+
pub async fn get_slice_logs(
294
+
pool: &PgPool,
295
+
slice_uri: &str,
296
+
log_type_filter: Option<&str>,
297
+
limit: Option<i64>,
298
+
) -> Result<Vec<LogEntry>, sqlx::Error> {
299
+
let limit = limit.unwrap_or(100);
300
+
301
+
let rows = if let Some(log_type) = log_type_filter {
302
+
sqlx::query_as!(
303
+
LogEntry,
304
+
r#"
305
+
SELECT id, created_at, log_type, job_id, user_did, slice_uri, level, message, metadata
306
+
FROM logs
307
+
WHERE slice_uri = $1 AND log_type = $2
308
+
ORDER BY created_at DESC
309
+
LIMIT $3
310
+
"#,
311
+
slice_uri,
312
+
log_type,
313
+
limit
314
+
)
315
+
.fetch_all(pool)
316
+
.await?
317
+
} else {
318
+
sqlx::query_as!(
319
+
LogEntry,
320
+
r#"
321
+
SELECT id, created_at, log_type, job_id, user_did, slice_uri, level, message, metadata
322
+
FROM logs
323
+
WHERE slice_uri = $1
324
+
ORDER BY created_at DESC
325
+
LIMIT $2
326
+
"#,
327
+
slice_uri,
328
+
limit
329
+
)
330
+
.fetch_all(pool)
331
+
.await?
332
+
};
333
+
334
+
Ok(rows)
335
+
}
336
+
337
+
/// Clean up old logs (keep last 30 days for jetstream, 7 days for completed sync jobs)
338
+
#[allow(dead_code)]
339
+
pub async fn cleanup_old_logs(pool: &PgPool) -> Result<u64, sqlx::Error> {
340
+
let result = sqlx::query!(
341
+
r#"
342
+
DELETE FROM logs
343
+
WHERE
344
+
(log_type = 'jetstream' AND created_at < NOW() - INTERVAL '30 days')
345
+
OR (log_type = 'sync_job' AND created_at < NOW() - INTERVAL '7 days')
346
+
OR (log_type = 'system' AND created_at < NOW() - INTERVAL '7 days')
347
+
"#,
348
+
)
349
+
.execute(pool)
350
+
.await?;
351
+
352
+
Ok(result.rows_affected())
353
+
}
+13
api/src/main.rs
+13
api/src/main.rs
···
7
7
mod handler_get_records;
8
8
mod handler_jetstream_status;
9
9
mod handler_jobs;
10
+
mod handler_logs;
10
11
mod handler_openapi_spec;
11
12
mod handler_stats;
12
13
mod handler_sync;
···
17
18
mod jetstream;
18
19
mod jobs;
19
20
mod lexicon;
21
+
mod logging;
20
22
mod models;
21
23
mod sync;
22
24
···
89
91
auth_base_url,
90
92
relay_endpoint,
91
93
};
94
+
95
+
// Initialize global logger
96
+
logging::Logger::init_global(pool.clone());
92
97
93
98
// Start job queue runner
94
99
let pool_for_runner = pool.clone();
···
296
301
.route(
297
302
"/xrpc/social.slices.slice.getJobHistory",
298
303
get(handler_jobs::get_slice_job_history),
304
+
)
305
+
.route(
306
+
"/xrpc/social.slices.slice.getJobLogs",
307
+
get(handler_logs::get_sync_job_logs_handler),
308
+
)
309
+
.route(
310
+
"/xrpc/social.slices.slice.getJetstreamLogs",
311
+
get(handler_logs::get_jetstream_logs_handler),
299
312
)
300
313
.route(
301
314
"/xrpc/social.slices.slice.stats",
+281
-59
api/src/sync.rs
+281
-59
api/src/sync.rs
···
14
14
use crate::errors::SyncError;
15
15
use crate::models::{Actor, Record};
16
16
use crate::lexicon::LexiconValidator;
17
+
use crate::logging::LogLevel;
18
+
use crate::logging::Logger;
19
+
use serde_json::json;
20
+
use uuid::Uuid;
17
21
18
22
19
23
#[derive(Debug, Deserialize)]
···
33
37
#[derive(Debug, Deserialize)]
34
38
struct ListReposByCollectionResponse {
35
39
repos: Vec<RepoRef>,
40
+
cursor: Option<String>,
36
41
}
37
42
38
43
#[derive(Debug, Deserialize)]
···
64
69
database: Database,
65
70
relay_endpoint: String,
66
71
atp_cache: std::sync::Arc<std::sync::Mutex<std::collections::HashMap<String, AtpData>>>,
72
+
logger: Option<Logger>,
73
+
job_id: Option<Uuid>,
74
+
user_did: Option<String>,
67
75
}
68
76
69
77
impl SyncService {
···
73
81
database,
74
82
relay_endpoint,
75
83
atp_cache: std::sync::Arc::new(std::sync::Mutex::new(std::collections::HashMap::new())),
84
+
logger: None,
85
+
job_id: None,
86
+
user_did: None,
76
87
}
77
88
}
78
89
90
+
/// Create a new SyncService with logging enabled for a specific job
91
+
pub fn with_logging(database: Database, relay_endpoint: String, logger: Logger, job_id: Uuid, user_did: String) -> Self {
92
+
Self {
93
+
client: Client::new(),
94
+
database,
95
+
relay_endpoint,
96
+
atp_cache: std::sync::Arc::new(std::sync::Mutex::new(std::collections::HashMap::new())),
97
+
logger: Some(logger),
98
+
job_id: Some(job_id),
99
+
user_did: Some(user_did),
100
+
}
101
+
}
79
102
103
+
/// Helper to log sync operations - stores slice_uri for reuse in the sync operation
104
+
fn log_with_context(&self, slice_uri: &str, level: LogLevel, message: &str, metadata: Option<serde_json::Value>) {
105
+
if let (Some(logger), Some(job_id), Some(user_did)) = (&self.logger, &self.job_id, &self.user_did) {
106
+
logger.log_sync_job(*job_id, user_did, slice_uri, level, message, metadata);
107
+
}
108
+
}
80
109
81
110
pub async fn backfill_collections(&self, slice_uri: &str, collections: Option<&[String]>, external_collections: Option<&[String]>, repos: Option<&[String]>, skip_validation: bool) -> Result<(i64, i64), SyncError> {
82
111
info!("🔄 Starting backfill operation");
···
112
141
// First, get all repos from primary collections
113
142
let mut primary_repos = std::collections::HashSet::new();
114
143
for collection in &primary_collections {
115
-
match self.get_repos_for_collection(collection).await {
144
+
match self.get_repos_for_collection(collection, slice_uri).await {
116
145
Ok(repos) => {
117
146
info!("✓ Found {} repositories for primary collection \"{}\"", repos.len(), collection);
147
+
self.log_with_context(slice_uri, LogLevel::Info,
148
+
&format!("Found {} repositories for collection '{}'", repos.len(), collection),
149
+
Some(json!({"collection": collection, "repo_count": repos.len()}))
150
+
);
118
151
primary_repos.extend(repos);
119
152
},
120
153
Err(e) => {
121
154
error!("Failed to get repos for primary collection {}: {}", collection, e);
155
+
self.log_with_context(slice_uri, LogLevel::Error,
156
+
&format!("Failed to fetch repositories for collection '{}': {}", collection, e),
157
+
Some(json!({"collection": collection, "error": e.to_string()}))
158
+
);
122
159
}
123
160
}
124
161
}
···
148
185
149
186
info!("🧠 Starting sync for {} repositories...", valid_repos.len());
150
187
151
-
// Create parallel fetch tasks for repos with valid ATP data only
188
+
// Group requests by PDS server to implement rate limiting
189
+
let mut requests_by_pds: std::collections::HashMap<String, Vec<(String, String)>> = std::collections::HashMap::new();
190
+
191
+
for repo in &valid_repos {
192
+
if let Some(atp_data) = atp_map.get(repo) {
193
+
let pds_url = atp_data.pds.clone();
194
+
for collection in &all_collections {
195
+
requests_by_pds
196
+
.entry(pds_url.clone())
197
+
.or_insert_with(Vec::new)
198
+
.push((repo.clone(), collection.clone()));
199
+
}
200
+
}
201
+
}
202
+
203
+
info!("📥 Fetching records with rate limiting: {} PDS servers, {} total requests",
204
+
requests_by_pds.len(),
205
+
requests_by_pds.values().map(|v| v.len()).sum::<usize>());
206
+
207
+
// Process each PDS server with limited concurrency
152
208
let mut fetch_tasks = Vec::new();
153
-
for repo in &valid_repos {
154
-
for collection in &all_collections {
155
-
let repo_clone = repo.clone();
156
-
let collection_clone = collection.clone();
157
-
let sync_service = self.clone();
158
-
let atp_map_clone = atp_map.clone();
159
-
let slice_uri_clone = slice_uri.to_string();
209
+
const MAX_CONCURRENT_PER_PDS: usize = 8; // Limit concurrent requests per PDS
160
210
161
-
let task = tokio::spawn(async move {
162
-
match sync_service.fetch_records_for_repo_collection_with_atp_map(&repo_clone, &collection_clone, &atp_map_clone, &slice_uri_clone).await {
163
-
Ok(records) => {
164
-
Ok((repo_clone, collection_clone, records))
165
-
}
166
-
Err(e) => {
167
-
// Handle common "not error" scenarios as empty results
168
-
match &e {
169
-
SyncError::ListRecords { status } => {
170
-
if *status == 404 || *status == 400 {
171
-
// Collection doesn't exist for this repo - return empty
172
-
Ok((repo_clone, collection_clone, vec![]))
173
-
} else {
174
-
Err(e)
175
-
}
211
+
for (_pds_url, repo_collections) in requests_by_pds {
212
+
let sync_service = self.clone();
213
+
let atp_map_clone = atp_map.clone();
214
+
let slice_uri_clone = slice_uri.to_string();
215
+
216
+
// Process this PDS server's requests in chunks
217
+
let pds_task = tokio::spawn(async move {
218
+
let mut pds_results = Vec::new();
219
+
220
+
// Split requests into chunks and process with limited concurrency
221
+
for chunk in repo_collections.chunks(MAX_CONCURRENT_PER_PDS) {
222
+
let mut chunk_tasks = Vec::new();
223
+
224
+
for (repo, collection) in chunk {
225
+
let repo_clone = repo.clone();
226
+
let collection_clone = collection.clone();
227
+
let sync_service_clone = sync_service.clone();
228
+
let atp_map_inner = atp_map_clone.clone();
229
+
let slice_uri_inner = slice_uri_clone.clone();
230
+
231
+
let task = tokio::spawn(async move {
232
+
match sync_service_clone.fetch_records_for_repo_collection_with_atp_map(&repo_clone, &collection_clone, &atp_map_inner, &slice_uri_inner).await {
233
+
Ok(records) => {
234
+
Ok((repo_clone, collection_clone, records))
176
235
}
177
-
SyncError::HttpRequest(_) => {
178
-
// Network errors - treat as empty (like TypeScript version)
179
-
Ok((repo_clone, collection_clone, vec![]))
236
+
Err(e) => {
237
+
// Handle common "not error" scenarios as empty results
238
+
match &e {
239
+
SyncError::ListRecords { status } => {
240
+
if *status == 404 || *status == 400 {
241
+
// Collection doesn't exist for this repo - return empty
242
+
Ok((repo_clone, collection_clone, vec![]))
243
+
} else {
244
+
Err(e)
245
+
}
246
+
}
247
+
SyncError::HttpRequest(_) => {
248
+
// Network errors - treat as empty (like TypeScript version)
249
+
Ok((repo_clone, collection_clone, vec![]))
250
+
}
251
+
_ => Err(e)
252
+
}
180
253
}
181
-
_ => Err(e)
182
254
}
255
+
});
256
+
chunk_tasks.push(task);
257
+
}
258
+
259
+
// Wait for this chunk to complete before starting the next
260
+
for task in chunk_tasks {
261
+
if let Ok(result) = task.await {
262
+
pds_results.push(result);
183
263
}
184
264
}
185
-
});
186
-
fetch_tasks.push(task);
187
-
}
188
-
}
189
265
190
-
info!("📥 Fetching records for repositories and collections...");
191
-
info!("🔧 Debug: Created {} fetch tasks for {} repos × {} collections", fetch_tasks.len(), valid_repos.len(), all_collections.len());
266
+
// Small delay between chunks to be kind to PDS servers
267
+
if chunk.len() == MAX_CONCURRENT_PER_PDS {
268
+
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
269
+
}
270
+
}
271
+
272
+
pds_results
273
+
});
274
+
275
+
fetch_tasks.push(pds_task);
276
+
}
192
277
193
278
// Collect all results
194
279
let mut all_records = Vec::new();
···
204
289
}
205
290
};
206
291
207
-
for task in fetch_tasks {
208
-
match task.await {
209
-
Ok(Ok((repo, collection, records))) => {
292
+
// Process results from each PDS server
293
+
for pds_task in fetch_tasks {
294
+
match pds_task.await {
295
+
Ok(pds_results) => {
296
+
// Process each result from this PDS
297
+
for result in pds_results {
298
+
match result {
299
+
Ok((repo, collection, records)) => {
210
300
let mut validated_records = Vec::new();
211
301
let total_records = records.len();
212
302
213
303
// Skip validation if requested
214
304
if skip_validation {
215
305
validated_records = records;
216
-
info!("⚠️ Validation skipped - accepting all {} records for collection {} from repo {}",
306
+
info!("⚠️ Validation skipped - accepting all {} records for collection {} from repo {}",
217
307
total_records, collection, repo);
218
308
}
219
309
// Validate each record if we have a validator
220
310
else if let Some(ref validator) = validator {
311
+
let mut validation_errors = Vec::new();
221
312
for record in records {
222
313
match validator.validate_record(&collection, &record.json) {
223
314
Ok(_) => {
224
315
validated_records.push(record);
225
316
}
226
317
Err(e) => {
227
-
warn!("Validation failed for record {} in collection {} from repo {}: {}",
228
-
record.uri, collection, repo, e);
318
+
let error_msg = format!("Validation failed for record {} from {}: {}", record.uri, repo, e);
319
+
warn!("{}", error_msg);
320
+
validation_errors.push(json!({
321
+
"uri": record.uri,
322
+
"error": e.to_string()
323
+
}));
324
+
325
+
// Log individual validation failures
326
+
self.log_with_context(slice_uri, LogLevel::Warn,
327
+
&error_msg,
328
+
Some(json!({
329
+
"repo": repo,
330
+
"collection": collection,
331
+
"record_uri": record.uri,
332
+
"validation_error": e.to_string()
333
+
}))
334
+
);
229
335
}
230
336
}
231
337
}
338
+
339
+
let valid_count = validated_records.len();
340
+
let invalid_count = validation_errors.len();
341
+
342
+
if invalid_count > 0 {
343
+
self.log_with_context(slice_uri, LogLevel::Warn,
344
+
&format!("Validation completed for {}/{}: {} valid, {} invalid records",
345
+
repo, collection, valid_count, invalid_count),
346
+
Some(json!({
347
+
"repo": repo,
348
+
"collection": collection,
349
+
"valid_records": valid_count,
350
+
"invalid_records": invalid_count,
351
+
"validation_errors": validation_errors
352
+
}))
353
+
);
354
+
} else {
355
+
self.log_with_context(slice_uri, LogLevel::Info,
356
+
&format!("All {} records validated successfully for {}/{}",
357
+
valid_count, repo, collection),
358
+
Some(json!({
359
+
"repo": repo,
360
+
"collection": collection,
361
+
"valid_records": valid_count
362
+
}))
363
+
);
364
+
}
365
+
232
366
info!("✓ Validated {}/{} records for collection {} from repo {}",
233
367
validated_records.len(), total_records, collection, repo);
234
368
} else {
235
369
// No validator available, accept all records
236
370
validated_records = records;
371
+
self.log_with_context(slice_uri, LogLevel::Warn,
372
+
&format!("No lexicon validator available for collection {}", collection),
373
+
Some(json!({"collection": collection, "repo": repo, "accepted_records": total_records}))
374
+
);
237
375
warn!("⚠️ No lexicon validator available - accepting all records without validation for collection {}", collection);
238
376
}
239
377
240
-
all_records.extend(validated_records);
241
-
successful_tasks += 1;
242
-
}
243
-
Ok(Err(_)) => {
244
-
failed_tasks += 1;
378
+
all_records.extend(validated_records);
379
+
successful_tasks += 1;
380
+
}
381
+
Err(_) => {
382
+
failed_tasks += 1;
383
+
}
384
+
}
385
+
}
245
386
}
246
-
Err(_e) => {
387
+
Err(_) => {
388
+
// PDS task failed - count all its requests as failed
247
389
failed_tasks += 1;
248
390
}
249
391
}
···
272
414
Ok((valid_repos.len() as i64, total_records))
273
415
}
274
416
275
-
async fn get_repos_for_collection(&self, collection: &str) -> Result<Vec<String>, SyncError> {
417
+
async fn get_repos_for_collection(&self, collection: &str, slice_uri: &str) -> Result<Vec<String>, SyncError> {
276
418
let url = format!("{}/xrpc/com.atproto.sync.listReposByCollection", self.relay_endpoint);
277
-
let response = self.client
278
-
.get(&url)
279
-
.query(&[("collection", collection)])
280
-
.send()
281
-
.await?;
419
+
let mut all_repos = Vec::new();
420
+
let mut cursor: Option<String> = None;
421
+
422
+
loop {
423
+
let mut query_params = vec![("collection", collection.to_string())];
424
+
if let Some(ref cursor_value) = cursor {
425
+
query_params.push(("cursor", cursor_value.clone()));
426
+
}
282
427
283
-
if !response.status().is_success() {
284
-
return Err(SyncError::ListRepos { status: response.status().as_u16() });
428
+
let response = self.client
429
+
.get(&url)
430
+
.query(&query_params)
431
+
.send()
432
+
.await?;
433
+
434
+
if !response.status().is_success() {
435
+
return Err(SyncError::ListRepos { status: response.status().as_u16() });
436
+
}
437
+
438
+
let repos_response: ListReposByCollectionResponse = response.json().await?;
439
+
440
+
// Add repos from this page to our collection
441
+
all_repos.extend(repos_response.repos.into_iter().map(|r| r.did));
442
+
443
+
// Check if there's a next page
444
+
match repos_response.cursor {
445
+
Some(next_cursor) if !next_cursor.is_empty() => {
446
+
cursor = Some(next_cursor);
447
+
// Log pagination progress if we have a logger
448
+
self.log_with_context(slice_uri, LogLevel::Info,
449
+
&format!("Fetching next page of repositories for collection {}, total so far: {}", collection, all_repos.len()),
450
+
Some(json!({
451
+
"collection": collection,
452
+
"repos_count": all_repos.len(),
453
+
"has_more": true
454
+
}))
455
+
);
456
+
}
457
+
_ => break, // No more pages
458
+
}
285
459
}
286
460
287
-
let repos_response: ListReposByCollectionResponse = response.json().await?;
288
-
Ok(repos_response.repos.into_iter().map(|r| r.did).collect())
461
+
// Log final count
462
+
self.log_with_context(slice_uri, LogLevel::Info,
463
+
&format!("Completed fetching repositories for collection {}, total: {}", collection, all_repos.len()),
464
+
Some(json!({
465
+
"collection": collection,
466
+
"total_repos": all_repos.len()
467
+
}))
468
+
);
469
+
470
+
Ok(all_repos)
289
471
}
290
472
291
473
async fn fetch_records_for_repo_collection_with_atp_map(&self, repo: &str, collection: &str, atp_map: &std::collections::HashMap<String, AtpData>, slice_uri: &str) -> Result<Vec<Record>, SyncError> {
···
312
494
params.push(("cursor", c));
313
495
}
314
496
497
+
let request_url = format!("{}/xrpc/com.atproto.repo.listRecords", pds_url);
315
498
let response = self.client
316
-
.get(&format!("{}/xrpc/com.atproto.repo.listRecords", pds_url))
499
+
.get(&request_url)
317
500
.query(¶ms)
318
501
.send()
319
-
.await?;
502
+
.await;
503
+
504
+
let response = match response {
505
+
Ok(resp) => resp,
506
+
Err(e) => {
507
+
self.log_with_context(slice_uri, LogLevel::Error,
508
+
&format!("Failed to fetch records from {}: Network error: {}", repo, e),
509
+
Some(json!({"repo": repo, "collection": collection, "pds_url": pds_url, "error": e.to_string()}))
510
+
);
511
+
return Err(SyncError::from(e));
512
+
}
513
+
};
320
514
321
515
if !response.status().is_success() {
322
-
return Err(SyncError::ListRecords { status: response.status().as_u16() });
516
+
let status = response.status().as_u16();
517
+
518
+
// HTTP 400/404 are expected when collections don't exist - log as info, not error
519
+
let (log_level, log_message) = if status == 400 || status == 404 {
520
+
(LogLevel::Info, format!("Collection '{}' not found for {}: HTTP {}", collection, repo, status))
521
+
} else {
522
+
(LogLevel::Error, format!("Failed to fetch records from {}: HTTP {} from PDS", repo, status))
523
+
};
524
+
525
+
self.log_with_context(slice_uri, log_level,
526
+
&log_message,
527
+
Some(json!({"repo": repo, "collection": collection, "pds_url": pds_url, "http_status": status}))
528
+
);
529
+
return Err(SyncError::ListRecords { status });
323
530
}
324
531
325
532
let list_response: ListRecordsResponse = response.json().await?;
···
352
559
if cursor.is_none() {
353
560
break;
354
561
}
562
+
}
563
+
564
+
// Log results for this repo/collection
565
+
if fetched_count > 0 || skipped_count > 0 {
566
+
self.log_with_context(slice_uri, LogLevel::Info,
567
+
&format!("Fetched {} new/changed, skipped {} unchanged records from {}/{}",
568
+
fetched_count, skipped_count, repo, collection),
569
+
Some(json!({
570
+
"repo": repo,
571
+
"collection": collection,
572
+
"new_records": fetched_count,
573
+
"skipped_records": skipped_count,
574
+
"pds_url": pds_url
575
+
}))
576
+
);
355
577
}
356
578
357
579
if skipped_count > 0 {
+232
-31
frontend/src/client.ts
+232
-31
frontend/src/client.ts
···
1
1
// Generated TypeScript client for AT Protocol records
2
-
// Generated at: 2025-09-03 03:15:34 UTC
2
+
// Generated at: 2025-09-03 22:34:38 UTC
3
3
// Lexicons: 6
4
4
5
5
/**
···
169
169
170
170
export type GetJobHistoryResponse = JobStatus[];
171
171
172
+
export interface GetJobLogsParams {
173
+
jobId: string;
174
+
limit?: number;
175
+
}
176
+
177
+
export interface GetJobLogsResponse {
178
+
logs: LogEntry[];
179
+
}
180
+
181
+
export interface LogEntry {
182
+
id: number;
183
+
createdAt: string;
184
+
logType: string;
185
+
jobId?: string;
186
+
userDid?: string;
187
+
sliceUri?: string;
188
+
level: string;
189
+
message: string;
190
+
metadata?: Record<string, unknown>;
191
+
}
192
+
172
193
export interface SyncUserCollectionsRequest {
173
194
slice: string;
174
195
timeoutSeconds?: number;
···
214
235
contains?: string;
215
236
}
216
237
217
-
export type WhereClause<T extends string = string> = Partial<
218
-
Record<T, WhereCondition>
219
-
> & { $or?: Partial<Record<T, WhereCondition>> };
238
+
export type WhereClause<T extends string = string> = {
239
+
[K in T]?: WhereCondition;
240
+
};
220
241
export type IndexedRecordFields =
221
242
| "did"
222
243
| "collection"
···
233
254
slice: string;
234
255
limit?: number;
235
256
cursor?: string;
236
-
where?: WhereClause<TSortField | IndexedRecordFields>;
257
+
where?: { [K in TSortField | IndexedRecordFields]?: WhereCondition };
258
+
orWhere?: { [K in TSortField | IndexedRecordFields]?: WhereCondition };
237
259
sortBy?: SortField<TSortField>[];
238
260
}
239
261
240
-
export interface SliceLevelRecordsParams {
262
+
export interface SliceLevelRecordsParams<TRecord = Record<string, unknown>> {
241
263
slice: string;
242
264
limit?: number;
243
265
cursor?: string;
244
-
where?: WhereClause<string>;
266
+
where?: { [K in keyof TRecord | IndexedRecordFields]?: WhereCondition };
267
+
orWhere?: { [K in keyof TRecord | IndexedRecordFields]?: WhereCondition };
245
268
sortBy?: SortField[];
246
269
}
247
270
···
545
568
async getRecords(params?: {
546
569
limit?: number;
547
570
cursor?: string;
548
-
where?: WhereClause<AppBskyActorProfileSortFields | IndexedRecordFields>;
571
+
where?: {
572
+
[K in
573
+
| AppBskyActorProfileSortFields
574
+
| IndexedRecordFields]?: WhereCondition;
575
+
};
576
+
orWhere?: {
577
+
[K in
578
+
| AppBskyActorProfileSortFields
579
+
| IndexedRecordFields]?: WhereCondition;
580
+
};
549
581
sortBy?: SortField<AppBskyActorProfileSortFields>[];
550
582
}): Promise<GetRecordsResponse<AppBskyActorProfile>> {
551
-
const requestParams = { ...params, slice: this.sliceUri };
583
+
// Combine where and orWhere into the expected backend format
584
+
const whereClause: any = params?.where ? { ...params.where } : {};
585
+
if (params?.orWhere) {
586
+
whereClause.$or = params.orWhere;
587
+
}
588
+
589
+
const requestParams = {
590
+
...params,
591
+
where: Object.keys(whereClause).length > 0 ? whereClause : undefined,
592
+
orWhere: undefined, // Remove orWhere as it's now in where.$or
593
+
slice: this.sliceUri,
594
+
};
552
595
const result = await this.makeRequest<SliceRecordsOutput>(
553
596
"app.bsky.actor.profile.getRecords",
554
597
"POST",
···
581
624
async countRecords(params?: {
582
625
limit?: number;
583
626
cursor?: string;
584
-
where?: WhereClause<AppBskyActorProfileSortFields | IndexedRecordFields>;
627
+
where?: {
628
+
[K in
629
+
| AppBskyActorProfileSortFields
630
+
| IndexedRecordFields]?: WhereCondition;
631
+
};
632
+
orWhere?: {
633
+
[K in
634
+
| AppBskyActorProfileSortFields
635
+
| IndexedRecordFields]?: WhereCondition;
636
+
};
585
637
sortBy?: SortField<AppBskyActorProfileSortFields>[];
586
638
}): Promise<CountRecordsResponse> {
587
-
const requestParams = { ...params, slice: this.sliceUri };
639
+
// Combine where and orWhere into the expected backend format
640
+
const whereClause: any = params?.where ? { ...params.where } : {};
641
+
if (params?.orWhere) {
642
+
whereClause.$or = params.orWhere;
643
+
}
644
+
645
+
const requestParams = {
646
+
...params,
647
+
where: Object.keys(whereClause).length > 0 ? whereClause : undefined,
648
+
orWhere: undefined, // Remove orWhere as it's now in where.$or
649
+
slice: this.sliceUri,
650
+
};
588
651
return await this.makeRequest<CountRecordsResponse>(
589
652
"app.bsky.actor.profile.countRecords",
590
653
"POST",
···
683
746
async getRecords(params?: {
684
747
limit?: number;
685
748
cursor?: string;
686
-
where?: WhereClause<SocialSlicesSliceSortFields | IndexedRecordFields>;
749
+
where?: {
750
+
[K in SocialSlicesSliceSortFields | IndexedRecordFields]?: WhereCondition;
751
+
};
752
+
orWhere?: {
753
+
[K in SocialSlicesSliceSortFields | IndexedRecordFields]?: WhereCondition;
754
+
};
687
755
sortBy?: SortField<SocialSlicesSliceSortFields>[];
688
756
}): Promise<GetRecordsResponse<SocialSlicesSlice>> {
689
-
const requestParams = { ...params, slice: this.sliceUri };
757
+
// Combine where and orWhere into the expected backend format
758
+
const whereClause: any = params?.where ? { ...params.where } : {};
759
+
if (params?.orWhere) {
760
+
whereClause.$or = params.orWhere;
761
+
}
762
+
763
+
const requestParams = {
764
+
...params,
765
+
where: Object.keys(whereClause).length > 0 ? whereClause : undefined,
766
+
orWhere: undefined, // Remove orWhere as it's now in where.$or
767
+
slice: this.sliceUri,
768
+
};
690
769
const result = await this.makeRequest<SliceRecordsOutput>(
691
770
"social.slices.slice.getRecords",
692
771
"POST",
···
719
798
async countRecords(params?: {
720
799
limit?: number;
721
800
cursor?: string;
722
-
where?: WhereClause<SocialSlicesSliceSortFields | IndexedRecordFields>;
801
+
where?: {
802
+
[K in SocialSlicesSliceSortFields | IndexedRecordFields]?: WhereCondition;
803
+
};
804
+
orWhere?: {
805
+
[K in SocialSlicesSliceSortFields | IndexedRecordFields]?: WhereCondition;
806
+
};
723
807
sortBy?: SortField<SocialSlicesSliceSortFields>[];
724
808
}): Promise<CountRecordsResponse> {
725
-
const requestParams = { ...params, slice: this.sliceUri };
809
+
// Combine where and orWhere into the expected backend format
810
+
const whereClause: any = params?.where ? { ...params.where } : {};
811
+
if (params?.orWhere) {
812
+
whereClause.$or = params.orWhere;
813
+
}
814
+
815
+
const requestParams = {
816
+
...params,
817
+
where: Object.keys(whereClause).length > 0 ? whereClause : undefined,
818
+
orWhere: undefined, // Remove orWhere as it's now in where.$or
819
+
slice: this.sliceUri,
820
+
};
726
821
return await this.makeRequest<CountRecordsResponse>(
727
822
"social.slices.slice.countRecords",
728
823
"POST",
···
789
884
}
790
885
791
886
async getSliceRecords<T = Record<string, unknown>>(
792
-
params: Omit<SliceLevelRecordsParams, "slice">
887
+
params: Omit<SliceLevelRecordsParams<T>, "slice">
793
888
): Promise<SliceRecordsOutput<T>> {
794
-
const requestParams = { ...params, slice: this.sliceUri };
889
+
// Combine where and orWhere into the expected backend format
890
+
const whereClause: any = params?.where ? { ...params.where } : {};
891
+
if (params?.orWhere) {
892
+
whereClause.$or = params.orWhere;
893
+
}
894
+
895
+
const requestParams = {
896
+
...params,
897
+
where: Object.keys(whereClause).length > 0 ? whereClause : undefined,
898
+
orWhere: undefined, // Remove orWhere as it's now in where.$or
899
+
slice: this.sliceUri,
900
+
};
795
901
return await this.makeRequest<SliceRecordsOutput<T>>(
796
902
"social.slices.slice.getSliceRecords",
797
903
"POST",
···
835
941
);
836
942
}
837
943
944
+
async getJobLogs(params: GetJobLogsParams): Promise<GetJobLogsResponse> {
945
+
return await this.makeRequest<GetJobLogsResponse>(
946
+
"social.slices.slice.getJobLogs",
947
+
"GET",
948
+
params
949
+
);
950
+
}
951
+
838
952
async getJetstreamStatus(): Promise<JetstreamStatusResponse> {
839
953
return await this.makeRequest<JetstreamStatusResponse>(
840
954
"social.slices.slice.getJetstreamStatus",
···
865
979
async getRecords(params?: {
866
980
limit?: number;
867
981
cursor?: string;
868
-
where?: WhereClause<SocialSlicesLexiconSortFields | IndexedRecordFields>;
982
+
where?: {
983
+
[K in
984
+
| SocialSlicesLexiconSortFields
985
+
| IndexedRecordFields]?: WhereCondition;
986
+
};
987
+
orWhere?: {
988
+
[K in
989
+
| SocialSlicesLexiconSortFields
990
+
| IndexedRecordFields]?: WhereCondition;
991
+
};
869
992
sortBy?: SortField<SocialSlicesLexiconSortFields>[];
870
993
}): Promise<GetRecordsResponse<SocialSlicesLexicon>> {
871
-
const requestParams = { ...params, slice: this.sliceUri };
994
+
// Combine where and orWhere into the expected backend format
995
+
const whereClause: any = params?.where ? { ...params.where } : {};
996
+
if (params?.orWhere) {
997
+
whereClause.$or = params.orWhere;
998
+
}
999
+
1000
+
const requestParams = {
1001
+
...params,
1002
+
where: Object.keys(whereClause).length > 0 ? whereClause : undefined,
1003
+
orWhere: undefined, // Remove orWhere as it's now in where.$or
1004
+
slice: this.sliceUri,
1005
+
};
872
1006
const result = await this.makeRequest<SliceRecordsOutput>(
873
1007
"social.slices.lexicon.getRecords",
874
1008
"POST",
···
901
1035
async countRecords(params?: {
902
1036
limit?: number;
903
1037
cursor?: string;
904
-
where?: WhereClause<SocialSlicesLexiconSortFields | IndexedRecordFields>;
1038
+
where?: {
1039
+
[K in
1040
+
| SocialSlicesLexiconSortFields
1041
+
| IndexedRecordFields]?: WhereCondition;
1042
+
};
1043
+
orWhere?: {
1044
+
[K in
1045
+
| SocialSlicesLexiconSortFields
1046
+
| IndexedRecordFields]?: WhereCondition;
1047
+
};
905
1048
sortBy?: SortField<SocialSlicesLexiconSortFields>[];
906
1049
}): Promise<CountRecordsResponse> {
907
-
const requestParams = { ...params, slice: this.sliceUri };
1050
+
// Combine where and orWhere into the expected backend format
1051
+
const whereClause: any = params?.where ? { ...params.where } : {};
1052
+
if (params?.orWhere) {
1053
+
whereClause.$or = params.orWhere;
1054
+
}
1055
+
1056
+
const requestParams = {
1057
+
...params,
1058
+
where: Object.keys(whereClause).length > 0 ? whereClause : undefined,
1059
+
orWhere: undefined, // Remove orWhere as it's now in where.$or
1060
+
slice: this.sliceUri,
1061
+
};
908
1062
return await this.makeRequest<CountRecordsResponse>(
909
1063
"social.slices.lexicon.countRecords",
910
1064
"POST",
···
966
1120
async getRecords(params?: {
967
1121
limit?: number;
968
1122
cursor?: string;
969
-
where?: WhereClause<
970
-
SocialSlicesActorProfileSortFields | IndexedRecordFields
971
-
>;
1123
+
where?: {
1124
+
[K in
1125
+
| SocialSlicesActorProfileSortFields
1126
+
| IndexedRecordFields]?: WhereCondition;
1127
+
};
1128
+
orWhere?: {
1129
+
[K in
1130
+
| SocialSlicesActorProfileSortFields
1131
+
| IndexedRecordFields]?: WhereCondition;
1132
+
};
972
1133
sortBy?: SortField<SocialSlicesActorProfileSortFields>[];
973
1134
}): Promise<GetRecordsResponse<SocialSlicesActorProfile>> {
974
-
const requestParams = { ...params, slice: this.sliceUri };
1135
+
// Combine where and orWhere into the expected backend format
1136
+
const whereClause: any = params?.where ? { ...params.where } : {};
1137
+
if (params?.orWhere) {
1138
+
whereClause.$or = params.orWhere;
1139
+
}
1140
+
1141
+
const requestParams = {
1142
+
...params,
1143
+
where: Object.keys(whereClause).length > 0 ? whereClause : undefined,
1144
+
orWhere: undefined, // Remove orWhere as it's now in where.$or
1145
+
slice: this.sliceUri,
1146
+
};
975
1147
const result = await this.makeRequest<SliceRecordsOutput>(
976
1148
"social.slices.actor.profile.getRecords",
977
1149
"POST",
···
1004
1176
async countRecords(params?: {
1005
1177
limit?: number;
1006
1178
cursor?: string;
1007
-
where?: WhereClause<
1008
-
SocialSlicesActorProfileSortFields | IndexedRecordFields
1009
-
>;
1179
+
where?: {
1180
+
[K in
1181
+
| SocialSlicesActorProfileSortFields
1182
+
| IndexedRecordFields]?: WhereCondition;
1183
+
};
1184
+
orWhere?: {
1185
+
[K in
1186
+
| SocialSlicesActorProfileSortFields
1187
+
| IndexedRecordFields]?: WhereCondition;
1188
+
};
1010
1189
sortBy?: SortField<SocialSlicesActorProfileSortFields>[];
1011
1190
}): Promise<CountRecordsResponse> {
1012
-
const requestParams = { ...params, slice: this.sliceUri };
1191
+
// Combine where and orWhere into the expected backend format
1192
+
const whereClause: any = params?.where ? { ...params.where } : {};
1193
+
if (params?.orWhere) {
1194
+
whereClause.$or = params.orWhere;
1195
+
}
1196
+
1197
+
const requestParams = {
1198
+
...params,
1199
+
where: Object.keys(whereClause).length > 0 ? whereClause : undefined,
1200
+
orWhere: undefined, // Remove orWhere as it's now in where.$or
1201
+
slice: this.sliceUri,
1202
+
};
1013
1203
return await this.makeRequest<CountRecordsResponse>(
1014
1204
"social.slices.actor.profile.countRecords",
1015
1205
"POST",
···
1129
1319
}
1130
1320
1131
1321
async getSliceRecords<T = Record<string, unknown>>(
1132
-
params: Omit<SliceLevelRecordsParams, "slice">
1322
+
params: Omit<SliceLevelRecordsParams<T>, "slice">
1133
1323
): Promise<SliceRecordsOutput<T>> {
1134
-
const requestParams = { ...params, slice: this.sliceUri };
1324
+
// Combine where and orWhere into the expected backend format
1325
+
const whereClause: any = params?.where ? { ...params.where } : {};
1326
+
if (params?.orWhere) {
1327
+
whereClause.$or = params.orWhere;
1328
+
}
1329
+
1330
+
const requestParams = {
1331
+
...params,
1332
+
where: Object.keys(whereClause).length > 0 ? whereClause : undefined,
1333
+
orWhere: undefined, // Remove orWhere as it's now in where.$or
1334
+
slice: this.sliceUri,
1335
+
};
1135
1336
return await this.makeRequest<SliceRecordsOutput<T>>(
1136
1337
"social.slices.slice.getSliceRecords",
1137
1338
"POST",
+12
-3
frontend/src/components/JobHistory.tsx
+12
-3
frontend/src/components/JobHistory.tsx
···
17
17
18
18
interface JobHistoryProps {
19
19
jobs: JobHistoryItem[];
20
+
sliceId: string;
20
21
}
21
22
22
23
function formatDate(dateString: string): string {
···
50
51
return `${value}${unit}`;
51
52
}
52
53
53
-
export function JobHistory({ jobs }: JobHistoryProps) {
54
+
export function JobHistory({ jobs, sliceId }: JobHistoryProps) {
54
55
if (jobs.length === 0) {
55
56
return (
56
57
<div className="bg-gray-50 border border-gray-200 rounded-lg p-6 text-center">
···
117
118
)}
118
119
</div>
119
120
120
-
<div className="text-xs text-gray-400 font-mono">
121
-
{job.jobId.split('-')[0]}...
121
+
<div className="flex flex-col items-end gap-2">
122
+
<div className="text-xs text-gray-400 font-mono">
123
+
{job.jobId.split('-')[0]}...
124
+
</div>
125
+
<a
126
+
href={`/slices/${sliceId}/sync/logs/${job.jobId}`}
127
+
className="text-xs text-blue-600 hover:text-blue-800 font-medium"
128
+
>
129
+
View Logs →
130
+
</a>
122
131
</div>
123
132
</div>
124
133
</div>
+129
frontend/src/components/SyncJobLogs.tsx
+129
frontend/src/components/SyncJobLogs.tsx
···
1
+
interface LogEntry {
2
+
id: number;
3
+
createdAt: string;
4
+
logType: string;
5
+
jobId?: string;
6
+
userDid?: string;
7
+
sliceUri?: string;
8
+
level: string;
9
+
message: string;
10
+
metadata?: Record<string, unknown>;
11
+
}
12
+
13
+
interface SyncJobLogsProps {
14
+
logs: LogEntry[];
15
+
jobId?: string;
16
+
}
17
+
18
+
function formatTimestamp(dateString: string): string {
19
+
const date = new Date(dateString);
20
+
return date.toLocaleTimeString([], {
21
+
hour: "2-digit",
22
+
minute: "2-digit",
23
+
second: "2-digit",
24
+
fractionalSecondDigits: 3,
25
+
});
26
+
}
27
+
28
+
function LogLevelBadge({ level }: { level: string }) {
29
+
const colors: Record<string, string> = {
30
+
error: "bg-red-100 text-red-800",
31
+
warn: "bg-yellow-100 text-yellow-800",
32
+
info: "bg-blue-100 text-blue-800",
33
+
debug: "bg-gray-100 text-gray-800",
34
+
};
35
+
36
+
return (
37
+
<span
38
+
className={`px-2 py-1 rounded text-xs font-medium ${
39
+
colors[level] || colors.debug
40
+
}`}
41
+
>
42
+
{level.toUpperCase()}
43
+
</span>
44
+
);
45
+
}
46
+
47
+
export function SyncJobLogs({ logs, jobId }: SyncJobLogsProps) {
48
+
if (logs.length === 0) {
49
+
return (
50
+
<div className="p-8 text-center text-gray-500">
51
+
No logs found for this job
52
+
{jobId && (
53
+
<div className="text-xs text-gray-400 mt-1 font-mono">
54
+
Job ID: {jobId}
55
+
</div>
56
+
)}
57
+
</div>
58
+
);
59
+
}
60
+
61
+
const errorCount = logs.filter((l) => l.level === "error").length;
62
+
const warnCount = logs.filter((l) => l.level === "warn").length;
63
+
const infoCount = logs.filter((l) => l.level === "info").length;
64
+
65
+
return (
66
+
<div className="divide-y divide-gray-200">
67
+
{/* Log Stats Header */}
68
+
<div className="p-4 bg-gray-50">
69
+
<div className="flex gap-4 text-sm">
70
+
<span>
71
+
Total logs: <strong>{logs.length}</strong>
72
+
</span>
73
+
{errorCount > 0 && (
74
+
<span className="text-red-600">
75
+
Errors: <strong>{errorCount}</strong>
76
+
</span>
77
+
)}
78
+
{warnCount > 0 && (
79
+
<span className="text-yellow-600">
80
+
Warnings: <strong>{warnCount}</strong>
81
+
</span>
82
+
)}
83
+
<span className="text-blue-600">
84
+
Info: <strong>{infoCount}</strong>
85
+
</span>
86
+
</div>
87
+
</div>
88
+
89
+
{/* Log Entries */}
90
+
<div className="max-h-[600px] overflow-y-auto">
91
+
{logs.map((log) => (
92
+
<div
93
+
key={log.id}
94
+
className={`p-3 hover:bg-gray-50 font-mono text-sm ${
95
+
log.level === "error"
96
+
? "bg-red-50"
97
+
: log.level === "warn"
98
+
? "bg-yellow-50"
99
+
: ""
100
+
}`}
101
+
>
102
+
<div className="flex items-start gap-3">
103
+
<span className="text-gray-400 text-xs">
104
+
{formatTimestamp(log.createdAt)}
105
+
</span>
106
+
<LogLevelBadge level={log.level} />
107
+
<div className="flex-1">
108
+
<div className="text-gray-800">{log.message}</div>
109
+
{log.metadata && Object.keys(log.metadata).length > 0 && (
110
+
<details className="mt-2">
111
+
<summary
112
+
className="text-xs text-gray-500 cursor-pointer hover:text-gray-700"
113
+
_="on click toggle .hidden on next <pre/>"
114
+
>
115
+
View metadata
116
+
</summary>
117
+
<pre className="mt-2 p-2 bg-gray-100 rounded text-xs overflow-x-auto hidden">
118
+
{JSON.stringify(log.metadata, null, 2)}
119
+
</pre>
120
+
</details>
121
+
)}
122
+
</div>
123
+
</div>
124
+
</div>
125
+
))}
126
+
</div>
127
+
</div>
128
+
);
129
+
}
+10
-6
frontend/src/pages/SliceSyncPage.tsx
+10
-6
frontend/src/pages/SliceSyncPage.tsx
···
84
84
: "Enter external collections (not matching your domain), one per line:\n\napp.bsky.feed.post\napp.bsky.actor.profile"
85
85
}
86
86
>
87
-
{externalCollections.length > 0 ? externalCollections.join("\n") : ""}
87
+
{externalCollections.length > 0
88
+
? externalCollections.join("\n")
89
+
: ""}
88
90
</textarea>
89
91
<p className="mt-1 text-xs text-gray-500">
90
-
External collections are those that don't match your slice's domain.
92
+
External collections are those that don't match your slice's
93
+
domain.
91
94
</p>
92
95
</div>
93
96
···
135
138
hx-swap="innerHTML"
136
139
className="mb-6"
137
140
>
138
-
<JobHistory jobs={[]} />
141
+
<JobHistory jobs={[]} sliceId={sliceId} />
139
142
</div>
140
143
141
144
<div className="bg-blue-50 border border-blue-200 rounded-lg p-6">
···
144
147
</h3>
145
148
<ul className="text-blue-700 space-y-1 text-sm">
146
149
<li>
147
-
• Primary collections matching your slice domain are automatically loaded
148
-
in the first field
150
+
• Primary collections matching your slice domain are automatically
151
+
loaded in the first field
149
152
</li>
150
153
<li>
151
-
• External collections from other domains are loaded in the second field
154
+
• External collections from other domains are loaded in the second
155
+
field
152
156
</li>
153
157
<li>
154
158
• Use External Collections to sync popular collections like{" "}
+48
frontend/src/pages/SyncJobLogsPage.tsx
+48
frontend/src/pages/SyncJobLogsPage.tsx
···
1
+
import { Layout } from "../components/Layout.tsx";
2
+
3
+
interface SyncJobLogsPageProps {
4
+
sliceName?: string;
5
+
sliceId?: string;
6
+
jobId?: string;
7
+
currentUser?: { handle?: string; isAuthenticated: boolean };
8
+
}
9
+
10
+
export function SyncJobLogsPage({
11
+
sliceId = "example",
12
+
jobId = "",
13
+
currentUser,
14
+
}: SyncJobLogsPageProps) {
15
+
return (
16
+
<Layout title={`Job Logs - ${jobId.substring(0, 8)}...`} currentUser={currentUser}>
17
+
<div>
18
+
<div className="flex items-center justify-between mb-8">
19
+
<div className="flex items-center">
20
+
<a
21
+
href={`/slices/${sliceId}/sync`}
22
+
className="text-blue-600 hover:text-blue-800 mr-4"
23
+
>
24
+
← Back to Sync
25
+
</a>
26
+
<h1 className="text-3xl font-bold text-gray-800">
27
+
Sync Job Logs
28
+
</h1>
29
+
</div>
30
+
<div className="text-sm text-gray-500 font-mono">
31
+
Job ID: {jobId}
32
+
</div>
33
+
</div>
34
+
35
+
<div
36
+
className="bg-white rounded-lg shadow-md"
37
+
hx-get={`/api/slices/${sliceId}/sync/logs/${jobId}`}
38
+
hx-trigger="load"
39
+
hx-swap="innerHTML"
40
+
>
41
+
<div className="p-8 text-center text-gray-500">
42
+
Loading logs...
43
+
</div>
44
+
</div>
45
+
</div>
46
+
</Layout>
47
+
);
48
+
}
+59
frontend/src/routes/pages.tsx
+59
frontend/src/routes/pages.tsx
···
13
13
import { SliceCodegenPage } from "../pages/SliceCodegenPage.tsx";
14
14
import { SliceApiDocsPage } from "../pages/SliceApiDocsPage.tsx";
15
15
import { SliceSettingsPage } from "../pages/SliceSettingsPage.tsx";
16
+
import { SyncJobLogsPage } from "../pages/SyncJobLogsPage.tsx";
16
17
import { SettingsPage } from "../pages/SettingsPage.tsx";
17
18
18
19
async function handleIndexPage(req: Request): Promise<Response> {
···
504
505
});
505
506
}
506
507
508
+
async function handleSyncJobLogsPage(
509
+
req: Request,
510
+
params?: URLPatternResult
511
+
): Promise<Response> {
512
+
const context = await withAuth(req);
513
+
514
+
if (!context.currentUser.isAuthenticated) {
515
+
return Response.redirect(new URL("/login", req.url), 302);
516
+
}
517
+
518
+
const sliceId = params?.pathname.groups.id;
519
+
const jobId = params?.pathname.groups.jobId;
520
+
521
+
if (!sliceId || !jobId) {
522
+
return new Response("Invalid slice ID or job ID", { status: 400 });
523
+
}
524
+
525
+
// Get slice details to pass slice name
526
+
let slice: { name: string } = { name: "Unknown Slice" };
527
+
try {
528
+
const sliceClient = getSliceClient(context, sliceId);
529
+
const sliceRecord = await sliceClient.social.slices.slice.getRecord({
530
+
uri: buildAtUri({
531
+
did: context.currentUser.sub!,
532
+
collection: "social.slices.slice",
533
+
rkey: sliceId,
534
+
}),
535
+
});
536
+
if (sliceRecord) {
537
+
slice = { name: sliceRecord.value.name };
538
+
}
539
+
} catch (error) {
540
+
console.error("Failed to fetch slice:", error);
541
+
}
542
+
543
+
const html = render(
544
+
<SyncJobLogsPage
545
+
sliceName={slice.name}
546
+
sliceId={sliceId}
547
+
jobId={jobId}
548
+
currentUser={context.currentUser}
549
+
/>
550
+
);
551
+
552
+
const responseHeaders: Record<string, string> = {
553
+
"content-type": "text/html",
554
+
};
555
+
556
+
return new Response(`<!DOCTYPE html>${html}`, {
557
+
status: 200,
558
+
headers: responseHeaders,
559
+
});
560
+
}
561
+
507
562
export const pageRoutes: Route[] = [
508
563
{
509
564
pattern: new URLPattern({ pathname: "/" }),
···
524
579
{
525
580
pattern: new URLPattern({ pathname: "/slices/:id/api-docs" }),
526
581
handler: handleSliceApiDocsPage,
582
+
},
583
+
{
584
+
pattern: new URLPattern({ pathname: "/slices/:id/sync/logs/:jobId" }),
585
+
handler: handleSyncJobLogsPage,
527
586
},
528
587
{
529
588
pattern: new URLPattern({ pathname: "/slices/:id/:tab" }),
+63
-2
frontend/src/routes/slices.tsx
+63
-2
frontend/src/routes/slices.tsx
···
17
17
import { SyncResult } from "../components/SyncResult.tsx";
18
18
import { JobHistory } from "../components/JobHistory.tsx";
19
19
import { JetstreamStatus } from "../components/JetstreamStatus.tsx";
20
+
import { SyncJobLogs } from "../components/SyncJobLogs.tsx";
20
21
import { buildAtUri } from "../utils/at-uri.ts";
21
22
22
23
async function handleCreateSlice(req: Request): Promise<Response> {
···
728
729
});
729
730
730
731
const jobs = result || [];
731
-
const html = render(<JobHistory jobs={jobs} />);
732
+
const html = render(<JobHistory jobs={jobs} sliceId={sliceId} />);
732
733
733
734
return new Response(html, {
734
735
status: 200,
···
736
737
});
737
738
} catch (error) {
738
739
console.error("Failed to get job history:", error);
739
-
const html = render(<JobHistory jobs={[]} />);
740
+
const html = render(<JobHistory jobs={[]} sliceId={sliceId} />);
741
+
return new Response(html, {
742
+
status: 200,
743
+
headers: { "content-type": "text/html" },
744
+
});
745
+
}
746
+
}
747
+
748
+
async function handleSyncJobLogs(
749
+
req: Request,
750
+
params?: URLPatternResult
751
+
): Promise<Response> {
752
+
const context = await withAuth(req);
753
+
const authResponse = requireAuth(context);
754
+
if (authResponse) return authResponse;
755
+
756
+
const sliceId = params?.pathname.groups.id;
757
+
const jobId = params?.pathname.groups.jobId;
758
+
759
+
if (!sliceId || !jobId) {
760
+
const html = render(
761
+
<div className="p-8 text-center text-red-600">
762
+
❌ Invalid slice ID or job ID
763
+
</div>
764
+
);
765
+
return new Response(html, {
766
+
status: 400,
767
+
headers: { "content-type": "text/html" },
768
+
});
769
+
}
770
+
771
+
try {
772
+
// Use the slice-specific client
773
+
const sliceClient = getSliceClient(context, sliceId);
774
+
775
+
// Get job logs
776
+
const result = await sliceClient.social.slices.slice.getJobLogs({
777
+
jobId: jobId,
778
+
limit: 1000,
779
+
});
780
+
781
+
const logs = result?.logs || [];
782
+
const html = render(<SyncJobLogs logs={logs} jobId={jobId} />);
783
+
740
784
return new Response(html, {
741
785
status: 200,
786
+
headers: { "content-type": "text/html" },
787
+
});
788
+
} catch (error) {
789
+
console.error("Failed to get sync job logs:", error);
790
+
const errorMessage = error instanceof Error ? error.message : String(error);
791
+
const html = render(
792
+
<div className="p-8 text-center text-red-600">
793
+
❌ Error loading logs: {errorMessage}
794
+
</div>
795
+
);
796
+
return new Response(html, {
797
+
status: 500,
742
798
headers: { "content-type": "text/html" },
743
799
});
744
800
}
···
917
973
method: "GET",
918
974
pattern: new URLPattern({ pathname: "/api/slices/:id/job-history" }),
919
975
handler: handleJobHistory,
976
+
},
977
+
{
978
+
method: "GET",
979
+
pattern: new URLPattern({ pathname: "/api/slices/:id/sync/logs/:jobId" }),
980
+
handler: handleSyncJobLogs,
920
981
},
921
982
{
922
983
method: "GET",