From cd4780af419d3a6de7401416c12789f0d50bcc6c Mon Sep 17 00:00:00 2001 From: lewis Date: Sun, 28 Dec 2025 23:41:17 +0200 Subject: [PATCH] Sync conformance fixes vs ref --- ...cc058ec6db0dde3ab0059d6f340a1fd9ddade.json | 29 -- ...1df772dc11ab682f47918cacb5ae186b4eb24.json | 25 - ...e91c4fa9753dc7c7fd964f6c43de27c5bac4a.json | 28 -- ...7b66211b8eb7ec5dadcfda79485520f792c16.json | 23 + ...0618a541ff9eafa457943ca814bd2a3ca72e1.json | 100 ++++ ...e3764e5761870aa8c10c50f1e3c8e7fdb87d4.json | 46 ++ ...762277849b75b647b7bf2ce5104baca795bf6.json | 29 ++ ...0a6a330f0cc8d72952d5b72a8766199fbf598.json | 20 + ...866bd099105461ffa126dcbe97ce9089a8b5d.json | 100 ++++ ...4c7c9959116bbdf4bed09939dfd82749ccb5f.json | 22 + ...e6c22308b878acdc5355dfcb04c5664c9463b.json | 34 ++ ...c35f7abc8590262156bfd4cdc4dbf11ddf8c9.json | 47 ++ justfile | 59 ++- src/api/admin/account/info.rs | 20 +- src/moderation/mod.rs | 4 +- src/sync/blob.rs | 89 ++-- src/sync/car.rs | 12 + src/sync/commit.rs | 165 ++++--- src/sync/frame.rs | 19 + src/sync/mod.rs | 4 + src/sync/repo.rs | 282 ++++++++--- src/sync/subscribe_repos.rs | 122 ++++- src/sync/util.rs | 180 ++++++- src/util.rs | 69 +++ tests/firehose_validation.rs | 199 +++++++- tests/helpers/mod.rs | 38 ++ tests/sync_blob.rs | 2 +- tests/sync_conformance.rs | 443 ++++++++++++++++++ tests/sync_deprecated.rs | 2 +- tests/sync_repo.rs | 67 +-- 30 files changed, 1928 insertions(+), 351 deletions(-) delete mode 100644 .sqlx/query-0f8fd9cbb1ff0fd8951ce082a82cc058ec6db0dde3ab0059d6f340a1fd9ddade.json delete mode 100644 .sqlx/query-0fdf13907693d130babae38f4bb1df772dc11ab682f47918cacb5ae186b4eb24.json delete mode 100644 .sqlx/query-1d3748694f23a407e26c793cc43e91c4fa9753dc7c7fd964f6c43de27c5bac4a.json create mode 100644 .sqlx/query-485cd286a085cca2910e3c3de757b66211b8eb7ec5dadcfda79485520f792c16.json create mode 100644 .sqlx/query-6783bd8e36444e5d6cc25cc1a120618a541ff9eafa457943ca814bd2a3ca72e1.json create mode 100644 .sqlx/query-93678a24667d311aaec7c6277aae3764e5761870aa8c10c50f1e3c8e7fdb87d4.json create mode 100644 .sqlx/query-9eeebac027c05ac44afa9f6b163762277849b75b647b7bf2ce5104baca795bf6.json create mode 100644 .sqlx/query-a805ece8ccc38c88a6dbca22dd70a6a330f0cc8d72952d5b72a8766199fbf598.json create mode 100644 .sqlx/query-abed6772d0cb2924c0aa27d479c866bd099105461ffa126dcbe97ce9089a8b5d.json create mode 100644 .sqlx/query-b43902272f2710b849840b29f2e4c7c9959116bbdf4bed09939dfd82749ccb5f.json create mode 100644 .sqlx/query-dd1b61d6ec81fd891d4effd3b51e6c22308b878acdc5355dfcb04c5664c9463b.json create mode 100644 .sqlx/query-f6723557ad451b8f4349df8ad4ec35f7abc8590262156bfd4cdc4dbf11ddf8c9.json create mode 100644 tests/sync_conformance.rs diff --git a/.sqlx/query-0f8fd9cbb1ff0fd8951ce082a82cc058ec6db0dde3ab0059d6f340a1fd9ddade.json b/.sqlx/query-0f8fd9cbb1ff0fd8951ce082a82cc058ec6db0dde3ab0059d6f340a1fd9ddade.json deleted file mode 100644 index 4552dba..0000000 --- a/.sqlx/query-0f8fd9cbb1ff0fd8951ce082a82cc058ec6db0dde3ab0059d6f340a1fd9ddade.json +++ /dev/null @@ -1,29 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n SELECT u.did, r.repo_root_cid\n FROM repos r\n JOIN users u ON r.user_id = u.id\n WHERE u.did > $1\n ORDER BY u.did ASC\n LIMIT $2\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "did", - "type_info": "Text" - }, - { - "ordinal": 1, - "name": "repo_root_cid", - "type_info": "Text" - } - ], - "parameters": { - "Left": [ - "Text", - "Int8" - ] - }, - "nullable": [ - false, - false - ] - }, - "hash": "0f8fd9cbb1ff0fd8951ce082a82cc058ec6db0dde3ab0059d6f340a1fd9ddade" -} diff --git a/.sqlx/query-0fdf13907693d130babae38f4bb1df772dc11ab682f47918cacb5ae186b4eb24.json b/.sqlx/query-0fdf13907693d130babae38f4bb1df772dc11ab682f47918cacb5ae186b4eb24.json deleted file mode 100644 index 0b0ae70..0000000 --- a/.sqlx/query-0fdf13907693d130babae38f4bb1df772dc11ab682f47918cacb5ae186b4eb24.json +++ /dev/null @@ -1,25 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n SELECT cid FROM blobs\n WHERE created_by_user = $1 AND cid > $2 AND created_at > $3\n ORDER BY cid ASC\n LIMIT $4\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "cid", - "type_info": "Text" - } - ], - "parameters": { - "Left": [ - "Uuid", - "Text", - "Timestamptz", - "Int8" - ] - }, - "nullable": [ - false - ] - }, - "hash": "0fdf13907693d130babae38f4bb1df772dc11ab682f47918cacb5ae186b4eb24" -} diff --git a/.sqlx/query-1d3748694f23a407e26c793cc43e91c4fa9753dc7c7fd964f6c43de27c5bac4a.json b/.sqlx/query-1d3748694f23a407e26c793cc43e91c4fa9753dc7c7fd964f6c43de27c5bac4a.json deleted file mode 100644 index 18577b9..0000000 --- a/.sqlx/query-1d3748694f23a407e26c793cc43e91c4fa9753dc7c7fd964f6c43de27c5bac4a.json +++ /dev/null @@ -1,28 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n SELECT u.did, r.repo_root_cid\n FROM users u\n LEFT JOIN repos r ON u.id = r.user_id\n WHERE u.did = $1\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "did", - "type_info": "Text" - }, - { - "ordinal": 1, - "name": "repo_root_cid", - "type_info": "Text" - } - ], - "parameters": { - "Left": [ - "Text" - ] - }, - "nullable": [ - false, - false - ] - }, - "hash": "1d3748694f23a407e26c793cc43e91c4fa9753dc7c7fd964f6c43de27c5bac4a" -} diff --git a/.sqlx/query-485cd286a085cca2910e3c3de757b66211b8eb7ec5dadcfda79485520f792c16.json b/.sqlx/query-485cd286a085cca2910e3c3de757b66211b8eb7ec5dadcfda79485520f792c16.json new file mode 100644 index 0000000..809b045 --- /dev/null +++ b/.sqlx/query-485cd286a085cca2910e3c3de757b66211b8eb7ec5dadcfda79485520f792c16.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT DISTINCT unnest(blobs) as \"cid!\"\n FROM repo_seq\n WHERE did = $1 AND rev > $2 AND blobs IS NOT NULL\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "cid!", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "485cd286a085cca2910e3c3de757b66211b8eb7ec5dadcfda79485520f792c16" +} diff --git a/.sqlx/query-6783bd8e36444e5d6cc25cc1a120618a541ff9eafa457943ca814bd2a3ca72e1.json b/.sqlx/query-6783bd8e36444e5d6cc25cc1a120618a541ff9eafa457943ca814bd2a3ca72e1.json new file mode 100644 index 0000000..205244e --- /dev/null +++ b/.sqlx/query-6783bd8e36444e5d6cc25cc1a120618a541ff9eafa457943ca814bd2a3ca72e1.json @@ -0,0 +1,100 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT seq, did, created_at, event_type, commit_cid, prev_cid, prev_data_cid, ops, blobs, blocks_cids, handle, active, status, rev\n FROM repo_seq\n WHERE seq > $1\n ORDER BY seq ASC\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "seq", + "type_info": "Int8" + }, + { + "ordinal": 1, + "name": "did", + "type_info": "Text" + }, + { + "ordinal": 2, + "name": "created_at", + "type_info": "Timestamptz" + }, + { + "ordinal": 3, + "name": "event_type", + "type_info": "Text" + }, + { + "ordinal": 4, + "name": "commit_cid", + "type_info": "Text" + }, + { + "ordinal": 5, + "name": "prev_cid", + "type_info": "Text" + }, + { + "ordinal": 6, + "name": "prev_data_cid", + "type_info": "Text" + }, + { + "ordinal": 7, + "name": "ops", + "type_info": "Jsonb" + }, + { + "ordinal": 8, + "name": "blobs", + "type_info": "TextArray" + }, + { + "ordinal": 9, + "name": "blocks_cids", + "type_info": "TextArray" + }, + { + "ordinal": 10, + "name": "handle", + "type_info": "Text" + }, + { + "ordinal": 11, + "name": "active", + "type_info": "Bool" + }, + { + "ordinal": 12, + "name": "status", + "type_info": "Text" + }, + { + "ordinal": 13, + "name": "rev", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [ + false, + false, + false, + false, + true, + true, + true, + true, + true, + true, + true, + true, + true, + true + ] + }, + "hash": "6783bd8e36444e5d6cc25cc1a120618a541ff9eafa457943ca814bd2a3ca72e1" +} diff --git a/.sqlx/query-93678a24667d311aaec7c6277aae3764e5761870aa8c10c50f1e3c8e7fdb87d4.json b/.sqlx/query-93678a24667d311aaec7c6277aae3764e5761870aa8c10c50f1e3c8e7fdb87d4.json new file mode 100644 index 0000000..c6c0ccd --- /dev/null +++ b/.sqlx/query-93678a24667d311aaec7c6277aae3764e5761870aa8c10c50f1e3c8e7fdb87d4.json @@ -0,0 +1,46 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT u.id, u.did, u.deactivated_at, u.takedown_ref, r.repo_root_cid\n FROM users u\n LEFT JOIN repos r ON r.user_id = u.id\n WHERE u.did = $1\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Uuid" + }, + { + "ordinal": 1, + "name": "did", + "type_info": "Text" + }, + { + "ordinal": 2, + "name": "deactivated_at", + "type_info": "Timestamptz" + }, + { + "ordinal": 3, + "name": "takedown_ref", + "type_info": "Text" + }, + { + "ordinal": 4, + "name": "repo_root_cid", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + false, + false, + true, + true, + false + ] + }, + "hash": "93678a24667d311aaec7c6277aae3764e5761870aa8c10c50f1e3c8e7fdb87d4" +} diff --git a/.sqlx/query-9eeebac027c05ac44afa9f6b163762277849b75b647b7bf2ce5104baca795bf6.json b/.sqlx/query-9eeebac027c05ac44afa9f6b163762277849b75b647b7bf2ce5104baca795bf6.json new file mode 100644 index 0000000..6424857 --- /dev/null +++ b/.sqlx/query-9eeebac027c05ac44afa9f6b163762277849b75b647b7bf2ce5104baca795bf6.json @@ -0,0 +1,29 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT blocks_cids, commit_cid\n FROM repo_seq\n WHERE did = $1 AND rev > $2\n ORDER BY seq DESC\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "blocks_cids", + "type_info": "TextArray" + }, + { + "ordinal": 1, + "name": "commit_cid", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + true, + true + ] + }, + "hash": "9eeebac027c05ac44afa9f6b163762277849b75b647b7bf2ce5104baca795bf6" +} diff --git a/.sqlx/query-a805ece8ccc38c88a6dbca22dd70a6a330f0cc8d72952d5b72a8766199fbf598.json b/.sqlx/query-a805ece8ccc38c88a6dbca22dd70a6a330f0cc8d72952d5b72a8766199fbf598.json new file mode 100644 index 0000000..debfae6 --- /dev/null +++ b/.sqlx/query-a805ece8ccc38c88a6dbca22dd70a6a330f0cc8d72952d5b72a8766199fbf598.json @@ -0,0 +1,20 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT MAX(seq) FROM repo_seq", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "max", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + null + ] + }, + "hash": "a805ece8ccc38c88a6dbca22dd70a6a330f0cc8d72952d5b72a8766199fbf598" +} diff --git a/.sqlx/query-abed6772d0cb2924c0aa27d479c866bd099105461ffa126dcbe97ce9089a8b5d.json b/.sqlx/query-abed6772d0cb2924c0aa27d479c866bd099105461ffa126dcbe97ce9089a8b5d.json new file mode 100644 index 0000000..1c1402e --- /dev/null +++ b/.sqlx/query-abed6772d0cb2924c0aa27d479c866bd099105461ffa126dcbe97ce9089a8b5d.json @@ -0,0 +1,100 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT seq, did, created_at, event_type, commit_cid, prev_cid, prev_data_cid, ops, blobs, blocks_cids, handle, active, status, rev\n FROM repo_seq\n WHERE seq > $1\n ORDER BY seq ASC\n LIMIT 1\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "seq", + "type_info": "Int8" + }, + { + "ordinal": 1, + "name": "did", + "type_info": "Text" + }, + { + "ordinal": 2, + "name": "created_at", + "type_info": "Timestamptz" + }, + { + "ordinal": 3, + "name": "event_type", + "type_info": "Text" + }, + { + "ordinal": 4, + "name": "commit_cid", + "type_info": "Text" + }, + { + "ordinal": 5, + "name": "prev_cid", + "type_info": "Text" + }, + { + "ordinal": 6, + "name": "prev_data_cid", + "type_info": "Text" + }, + { + "ordinal": 7, + "name": "ops", + "type_info": "Jsonb" + }, + { + "ordinal": 8, + "name": "blobs", + "type_info": "TextArray" + }, + { + "ordinal": 9, + "name": "blocks_cids", + "type_info": "TextArray" + }, + { + "ordinal": 10, + "name": "handle", + "type_info": "Text" + }, + { + "ordinal": 11, + "name": "active", + "type_info": "Bool" + }, + { + "ordinal": 12, + "name": "status", + "type_info": "Text" + }, + { + "ordinal": 13, + "name": "rev", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [ + false, + false, + false, + false, + true, + true, + true, + true, + true, + true, + true, + true, + true, + true + ] + }, + "hash": "abed6772d0cb2924c0aa27d479c866bd099105461ffa126dcbe97ce9089a8b5d" +} diff --git a/.sqlx/query-b43902272f2710b849840b29f2e4c7c9959116bbdf4bed09939dfd82749ccb5f.json b/.sqlx/query-b43902272f2710b849840b29f2e4c7c9959116bbdf4bed09939dfd82749ccb5f.json new file mode 100644 index 0000000..8a41f4c --- /dev/null +++ b/.sqlx/query-b43902272f2710b849840b29f2e4c7c9959116bbdf4bed09939dfd82749ccb5f.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT MIN(seq) FROM repo_seq WHERE created_at >= $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "min", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Timestamptz" + ] + }, + "nullable": [ + null + ] + }, + "hash": "b43902272f2710b849840b29f2e4c7c9959116bbdf4bed09939dfd82749ccb5f" +} diff --git a/.sqlx/query-dd1b61d6ec81fd891d4effd3b51e6c22308b878acdc5355dfcb04c5664c9463b.json b/.sqlx/query-dd1b61d6ec81fd891d4effd3b51e6c22308b878acdc5355dfcb04c5664c9463b.json new file mode 100644 index 0000000..236230f --- /dev/null +++ b/.sqlx/query-dd1b61d6ec81fd891d4effd3b51e6c22308b878acdc5355dfcb04c5664c9463b.json @@ -0,0 +1,34 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT storage_key, mime_type, size_bytes FROM blobs WHERE cid = $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "storage_key", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "mime_type", + "type_info": "Text" + }, + { + "ordinal": 2, + "name": "size_bytes", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + false, + false, + false + ] + }, + "hash": "dd1b61d6ec81fd891d4effd3b51e6c22308b878acdc5355dfcb04c5664c9463b" +} diff --git a/.sqlx/query-f6723557ad451b8f4349df8ad4ec35f7abc8590262156bfd4cdc4dbf11ddf8c9.json b/.sqlx/query-f6723557ad451b8f4349df8ad4ec35f7abc8590262156bfd4cdc4dbf11ddf8c9.json new file mode 100644 index 0000000..4485231 --- /dev/null +++ b/.sqlx/query-f6723557ad451b8f4349df8ad4ec35f7abc8590262156bfd4cdc4dbf11ddf8c9.json @@ -0,0 +1,47 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT u.did, u.deactivated_at, u.takedown_ref, r.repo_root_cid, r.repo_rev\n FROM repos r\n JOIN users u ON r.user_id = u.id\n WHERE u.did > $1\n ORDER BY u.did ASC\n LIMIT $2\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "did", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "deactivated_at", + "type_info": "Timestamptz" + }, + { + "ordinal": 2, + "name": "takedown_ref", + "type_info": "Text" + }, + { + "ordinal": 3, + "name": "repo_root_cid", + "type_info": "Text" + }, + { + "ordinal": 4, + "name": "repo_rev", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Text", + "Int8" + ] + }, + "nullable": [ + false, + true, + true, + false, + true + ] + }, + "hash": "f6723557ad451b8f4349df8ad4ec35f7abc8590262156bfd4cdc4dbf11ddf8c9" +} diff --git a/justfile b/justfile index 0ab8088..f9260b9 100644 --- a/justfile +++ b/justfile @@ -1,5 +1,6 @@ default: @just --list + run: cargo run run-release: @@ -17,22 +18,50 @@ fmt: fmt-check: cargo fmt -- --check lint: fmt-check clippy -# Run tests (auto-starts and auto-cleans containers) + +test-all *args: + ./scripts/run-tests.sh {{args}} + +test-auth: + ./scripts/run-tests.sh --test oauth --test oauth_lifecycle --test oauth_scopes --test oauth_security --test oauth_client_metadata --test jwt_security --test session_management --test change_password --test password_reset + +test-admin: + ./scripts/run-tests.sh --test admin_email --test admin_invite --test admin_moderation --test admin_search --test admin_stats + +test-sync: + ./scripts/run-tests.sh --test sync_repo --test sync_blob --test sync_conformance --test sync_deprecated --test firehose_validation + +test-repo: + ./scripts/run-tests.sh --test repo_batch --test repo_blob --test record_validation --test lifecycle_record + +test-identity: + ./scripts/run-tests.sh --test identity --test did_web --test plc_migration --test plc_operations --test plc_validation + +test-account: + ./scripts/run-tests.sh --test lifecycle_session --test delete_account --test invite --test email_update --test account_notifications + +test-security: + ./scripts/run-tests.sh --test security_fixes --test banned_words --test rate_limit --test moderation + +test-import: + ./scripts/run-tests.sh --test import_verification --test import_with_verification + +test-misc: + ./scripts/run-tests.sh --test actor --test commit_signing --test image_processing --test lifecycle_social --test notifications --test server --test signing_key --test verify_live_commit + test *args: ./scripts/run-tests.sh {{args}} -# Run a specific test file -test-file file: - ./scripts/run-tests.sh --test {{file}} -# Run tests with testcontainers (slower, no shared infra) -test-standalone: - TRANQUIL_PDS_ALLOW_INSECURE_SECRETS=1 cargo test -# Manually manage test infrastructure (for debugging) -test-infra-start: + +test-one name: + ./scripts/run-tests.sh --test {{name}} + +infra-start: ./scripts/test-infra.sh start -test-infra-stop: +infra-stop: ./scripts/test-infra.sh stop -test-infra-status: +infra-status: ./scripts/test-infra.sh status + clean: cargo clean doc: @@ -53,14 +82,14 @@ podman-logs: podman compose logs -f podman-build: podman compose build -# Frontend commands (Deno) + frontend-dev: . ~/.deno/env && cd frontend && deno task dev frontend-build: . ~/.deno/env && cd frontend && deno task build frontend-clean: rm -rf frontend/dist frontend/node_modules -# Frontend tests + frontend-test *args: . ~/.deno/env && cd frontend && VITEST=true deno task test:run {{args}} frontend-test-watch: @@ -69,7 +98,5 @@ frontend-test-ui: . ~/.deno/env && cd frontend && VITEST=true deno task test:ui frontend-test-coverage: . ~/.deno/env && cd frontend && VITEST=true deno task test:run --coverage -# Build all (frontend + backend) + build-all: frontend-build build -# Test all (backend + frontend) -test-all: test frontend-test diff --git a/src/api/admin/account/info.rs b/src/api/admin/account/info.rs index ff2e83d..2c45baf 100644 --- a/src/api/admin/account/info.rs +++ b/src/api/admin/account/info.rs @@ -88,30 +88,12 @@ pub async fn get_account_info( } } -fn parse_repeated_param(query: Option<&str>, key: &str) -> Vec { - query - .map(|q| { - q.split('&') - .filter_map(|pair| { - let (k, v) = pair.split_once('=')?; - - if k == key { - Some(urlencoding::decode(v).ok()?.into_owned()) - } else { - None - } - }) - .collect() - }) - .unwrap_or_default() -} - pub async fn get_account_infos( State(state): State, _auth: BearerAuthAdmin, RawQuery(raw_query): RawQuery, ) -> Response { - let dids = parse_repeated_param(raw_query.as_deref(), "dids"); + let dids = crate::util::parse_repeated_query_param(raw_query.as_deref(), "dids"); if dids.is_empty() { return ( StatusCode::BAD_REQUEST, diff --git a/src/moderation/mod.rs b/src/moderation/mod.rs index e73b663..fc02c31 100644 --- a/src/moderation/mod.rs +++ b/src/moderation/mod.rs @@ -107,7 +107,9 @@ mod tests { use base64::Engine; fn d(b64: &str) -> String { - let bytes = base64::engine::general_purpose::STANDARD.decode(b64).unwrap(); + let bytes = base64::engine::general_purpose::STANDARD + .decode(b64) + .unwrap(); String::from_utf8(bytes).unwrap() } diff --git a/src/sync/blob.rs b/src/sync/blob.rs index 4adeb1e..0043d71 100644 --- a/src/sync/blob.rs +++ b/src/sync/blob.rs @@ -1,4 +1,5 @@ use crate::state::AppState; +use crate::sync::util::assert_repo_availability; use axum::{ Json, body::Body, @@ -37,29 +38,14 @@ pub async fn get_blob( ) .into_response(); } - let user_exists = sqlx::query!("SELECT id FROM users WHERE did = $1", did) - .fetch_optional(&state.db) - .await; - match user_exists { - Ok(None) => { - return ( - StatusCode::NOT_FOUND, - Json(json!({"error": "RepoNotFound", "message": "Could not find repo for DID"})), - ) - .into_response(); - } - Err(e) => { - error!("DB error in get_blob: {:?}", e); - return ( - StatusCode::INTERNAL_SERVER_ERROR, - Json(json!({"error": "InternalError"})), - ) - .into_response(); - } - Ok(Some(_)) => {} - } + + let _account = match assert_repo_availability(&state.db, did, false).await { + Ok(a) => a, + Err(e) => return e.into_response(), + }; + let blob_result = sqlx::query!( - "SELECT storage_key, mime_type FROM blobs WHERE cid = $1", + "SELECT storage_key, mime_type, size_bytes FROM blobs WHERE cid = $1", cid ) .fetch_optional(&state.db) @@ -68,10 +54,14 @@ pub async fn get_blob( Ok(Some(row)) => { let storage_key = &row.storage_key; let mime_type = &row.mime_type; + let size_bytes = row.size_bytes; match state.blob_store.get(storage_key).await { Ok(data) => Response::builder() .status(StatusCode::OK) .header(header::CONTENT_TYPE, mime_type) + .header(header::CONTENT_LENGTH, size_bytes.to_string()) + .header("x-content-type-options", "nosniff") + .header("content-security-policy", "default-src 'none'; sandbox") .body(Body::from(data)) .unwrap(), Err(e) => { @@ -127,48 +117,35 @@ pub async fn list_blobs( ) .into_response(); } + + let account = match assert_repo_availability(&state.db, did, false).await { + Ok(a) => a, + Err(e) => return e.into_response(), + }; + let limit = params.limit.unwrap_or(500).clamp(1, 1000); let cursor_cid = params.cursor.as_deref().unwrap_or(""); - let user_result = sqlx::query!("SELECT id FROM users WHERE did = $1", did) - .fetch_optional(&state.db) - .await; - let user_id = match user_result { - Ok(Some(row)) => row.id, - Ok(None) => { - return ( - StatusCode::NOT_FOUND, - Json(json!({"error": "RepoNotFound", "message": "Could not find repo for DID"})), - ) - .into_response(); - } - Err(e) => { - error!("DB error in list_blobs: {:?}", e); - return ( - StatusCode::INTERNAL_SERVER_ERROR, - Json(json!({"error": "InternalError"})), - ) - .into_response(); - } - }; + let user_id = account.user_id; + let cids_result: Result, sqlx::Error> = if let Some(since) = ¶ms.since { - let since_time = chrono::DateTime::parse_from_rfc3339(since) - .map(|dt| dt.with_timezone(&chrono::Utc)) - .unwrap_or_else(|_| chrono::Utc::now()); - sqlx::query!( + sqlx::query_scalar!( r#" - SELECT cid FROM blobs - WHERE created_by_user = $1 AND cid > $2 AND created_at > $3 - ORDER BY cid ASC - LIMIT $4 + SELECT DISTINCT unnest(blobs) as "cid!" + FROM repo_seq + WHERE did = $1 AND rev > $2 AND blobs IS NOT NULL "#, - user_id, - cursor_cid, - since_time, - limit + 1 + did, + since ) .fetch_all(&state.db) .await - .map(|rows| rows.into_iter().map(|r| r.cid).collect()) + .map(|mut cids| { + cids.sort(); + cids.into_iter() + .filter(|c| c.as_str() > cursor_cid) + .take((limit + 1) as usize) + .collect() + }) } else { sqlx::query!( r#" diff --git a/src/sync/car.rs b/src/sync/car.rs index 7954e1e..774c04e 100644 --- a/src/sync/car.rs +++ b/src/sync/car.rs @@ -34,3 +34,15 @@ pub fn encode_car_header(root_cid: &Cid) -> Result, String> { result.extend_from_slice(&header_cbor); Ok(result) } + +pub fn encode_car_header_null_root() -> Result, String> { + let header = CarHeader::new_v1(vec![]); + let header_cbor = header + .encode() + .map_err(|e| format!("Failed to encode CAR header: {:?}", e))?; + let mut result = Vec::new(); + write_varint(&mut result, header_cbor.len() as u64) + .expect("Writing to Vec should never fail"); + result.extend_from_slice(&header_cbor); + Ok(result) +} diff --git a/src/sync/commit.rs b/src/sync/commit.rs index 9322d39..68dab6e 100644 --- a/src/sync/commit.rs +++ b/src/sync/commit.rs @@ -1,4 +1,5 @@ use crate::state::AppState; +use crate::sync::util::{AccountStatus, assert_repo_availability, get_account_with_status}; use axum::{ Json, extract::{Query, State}, @@ -43,45 +44,46 @@ pub async fn get_latest_commit( ) .into_response(); } - let result = sqlx::query!( - r#" - SELECT r.repo_root_cid - FROM repos r - JOIN users u ON r.user_id = u.id - WHERE u.did = $1 - "#, - did - ) - .fetch_optional(&state.db) - .await; - match result { - Ok(Some(row)) => { - let rev = get_rev_from_commit(&state, &row.repo_root_cid) - .await - .unwrap_or_else(|| chrono::Utc::now().timestamp_millis().to_string()); - ( - StatusCode::OK, - Json(GetLatestCommitOutput { - cid: row.repo_root_cid, - rev, - }), + + let account = match assert_repo_availability(&state.db, did, false).await { + Ok(a) => a, + Err(e) => return e.into_response(), + }; + + let repo_root_cid = match account.repo_root_cid { + Some(cid) => cid, + None => { + return ( + StatusCode::BAD_REQUEST, + Json(json!({"error": "RepoNotFound", "message": "Repo not initialized"})), ) - .into_response() + .into_response(); } - Ok(None) => ( - StatusCode::NOT_FOUND, - Json(json!({"error": "RepoNotFound", "message": "Could not find repo for DID"})), - ) - .into_response(), - Err(e) => { - error!("DB error in get_latest_commit: {:?}", e); - ( + }; + + let rev = match get_rev_from_commit(&state, &repo_root_cid).await { + Some(r) => r, + None => { + error!( + "Failed to parse commit for DID {}: CID {}", + did, repo_root_cid + ); + return ( StatusCode::INTERNAL_SERVER_ERROR, - Json(json!({"error": "InternalError"})), + Json(json!({"error": "InternalError", "message": "Failed to read repo commit"})), ) - .into_response() + .into_response(); } - } + }; + + ( + StatusCode::OK, + Json(GetLatestCommitOutput { + cid: repo_root_cid, + rev, + }), + ) + .into_response() } #[derive(Deserialize)] @@ -97,6 +99,8 @@ pub struct RepoInfo { pub head: String, pub rev: String, pub active: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub status: Option, } #[derive(Serialize)] @@ -114,7 +118,7 @@ pub async fn list_repos( let cursor_did = params.cursor.as_deref().unwrap_or(""); let result = sqlx::query!( r#" - SELECT u.did, r.repo_root_cid + SELECT u.did, u.deactivated_at, u.takedown_ref, r.repo_root_cid, r.repo_rev FROM repos r JOIN users u ON r.user_id = u.id WHERE u.did > $1 @@ -131,14 +135,34 @@ pub async fn list_repos( let has_more = rows.len() as i64 > limit; let mut repos: Vec = Vec::new(); for row in rows.iter().take(limit as usize) { - let rev = get_rev_from_commit(&state, &row.repo_root_cid) - .await - .unwrap_or_else(|| chrono::Utc::now().timestamp_millis().to_string()); + let rev = match get_rev_from_commit(&state, &row.repo_root_cid).await { + Some(r) => r, + None => { + if let Some(ref stored_rev) = row.repo_rev { + stored_rev.clone() + } else { + tracing::warn!( + "Failed to parse commit for DID {} in list_repos: CID {}", + row.did, + row.repo_root_cid + ); + continue; + } + } + }; + let status = if row.takedown_ref.is_some() { + AccountStatus::Takendown + } else if row.deactivated_at.is_some() { + AccountStatus::Deactivated + } else { + AccountStatus::Active + }; repos.push(RepoInfo { did: row.did.clone(), head: row.repo_root_cid.clone(), rev, - active: true, + active: status.is_active(), + status: status.as_str().map(String::from), }); } let next_cursor = if has_more { @@ -175,6 +199,9 @@ pub struct GetRepoStatusParams { pub struct GetRepoStatusOutput { pub did: String, pub active: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub status: Option, + #[serde(skip_serializing_if = "Option::is_none")] pub rev: Option, } @@ -190,42 +217,44 @@ pub async fn get_repo_status( ) .into_response(); } - let result = sqlx::query!( - r#" - SELECT u.did, r.repo_root_cid - FROM users u - LEFT JOIN repos r ON u.id = r.user_id - WHERE u.did = $1 - "#, - did - ) - .fetch_optional(&state.db) - .await; - match result { - Ok(Some(row)) => { - let rev = get_rev_from_commit(&state, &row.repo_root_cid).await; - ( - StatusCode::OK, - Json(GetRepoStatusOutput { - did: row.did, - active: true, - rev, - }), + + let account = match get_account_with_status(&state.db, did).await { + Ok(Some(a)) => a, + Ok(None) => { + return ( + StatusCode::BAD_REQUEST, + Json(json!({"error": "RepoNotFound", "message": format!("Could not find repo for DID: {}", did)})), ) .into_response() } - Ok(None) => ( - StatusCode::NOT_FOUND, - Json(json!({"error": "RepoNotFound", "message": "Could not find repo for DID"})), - ) - .into_response(), Err(e) => { error!("DB error in get_repo_status: {:?}", e); - ( + return ( StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"})), ) - .into_response() + .into_response(); } - } + }; + + let rev = if account.status.is_active() { + if let Some(ref cid) = account.repo_root_cid { + get_rev_from_commit(&state, cid).await + } else { + None + } + } else { + None + }; + + ( + StatusCode::OK, + Json(GetRepoStatusOutput { + did: account.did, + active: account.status.is_active(), + status: account.status.as_str().map(String::from), + rev, + }), + ) + .into_response() } diff --git a/src/sync/frame.rs b/src/sync/frame.rs index 8a613cd..13176a8 100644 --- a/src/sync/frame.rs +++ b/src/sync/frame.rs @@ -74,6 +74,25 @@ pub struct SyncFrame { pub time: String, } +#[derive(Debug, Serialize, Deserialize)] +pub struct InfoFrame { + pub name: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub message: Option, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct ErrorFrameHeader { + pub op: i64, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct ErrorFrameBody { + pub error: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub message: Option, +} + pub struct CommitFrameBuilder { pub seq: i64, pub did: String, diff --git a/src/sync/mod.rs b/src/sync/mod.rs index a0c2b43..a97860b 100644 --- a/src/sync/mod.rs +++ b/src/sync/mod.rs @@ -18,4 +18,8 @@ pub use crawl::{notify_of_update, request_crawl}; pub use deprecated::{get_checkout, get_head}; pub use repo::{get_blocks, get_record, get_repo}; pub use subscribe_repos::subscribe_repos; +pub use util::{ + AccountStatus, RepoAccount, RepoAvailabilityError, assert_repo_availability, + get_account_with_status, +}; pub use verify::{CarVerifier, VerifiedCar, VerifyError}; diff --git a/src/sync/repo.rs b/src/sync/repo.rs index fc4c78e..a26dffb 100644 --- a/src/sync/repo.rs +++ b/src/sync/repo.rs @@ -1,8 +1,9 @@ use crate::state::AppState; use crate::sync::car::encode_car_header; +use crate::sync::util::assert_repo_availability; use axum::{ Json, - extract::{Query, State}, + extract::{Query, RawQuery, State}, http::StatusCode, response::{IntoResponse, Response}, }; @@ -17,48 +18,102 @@ use tracing::error; const MAX_REPO_BLOCKS_TRAVERSAL: usize = 20_000; -#[derive(Deserialize)] -pub struct GetBlocksQuery { - pub did: String, - pub cids: String, +fn parse_get_blocks_query(query_string: &str) -> Result<(String, Vec), String> { + let did = crate::util::parse_repeated_query_param(Some(query_string), "did") + .into_iter() + .next() + .ok_or("Missing required parameter: did")?; + let cids = crate::util::parse_repeated_query_param(Some(query_string), "cids"); + Ok((did, cids)) } -pub async fn get_blocks( - State(state): State, - Query(query): Query, -) -> Response { - let user_exists = sqlx::query!("SELECT id FROM users WHERE did = $1", query.did) - .fetch_optional(&state.db) - .await - .unwrap_or(None); - if user_exists.is_none() { - return (StatusCode::NOT_FOUND, "Repo not found").into_response(); - } - let cids_str: Vec<&str> = query.cids.split(',').collect(); +pub async fn get_blocks(State(state): State, RawQuery(query): RawQuery) -> Response { + let query_string = match query { + Some(q) => q, + None => { + return ( + StatusCode::BAD_REQUEST, + Json(json!({"error": "InvalidRequest", "message": "Missing query parameters"})), + ) + .into_response(); + } + }; + + let (did, cid_strings) = match parse_get_blocks_query(&query_string) { + Ok(parsed) => parsed, + Err(msg) => { + return ( + StatusCode::BAD_REQUEST, + Json(json!({"error": "InvalidRequest", "message": msg})), + ) + .into_response(); + } + }; + + let _account = match assert_repo_availability(&state.db, &did, false).await { + Ok(a) => a, + Err(e) => return e.into_response(), + }; + let mut cids = Vec::new(); - for s in cids_str { + for s in &cid_strings { match Cid::from_str(s) { Ok(cid) => cids.push(cid), - Err(_) => return (StatusCode::BAD_REQUEST, "Invalid CID").into_response(), + Err(_) => return ( + StatusCode::BAD_REQUEST, + Json(json!({"error": "InvalidRequest", "message": format!("Invalid CID: {}", s)})), + ) + .into_response(), } } + + if cids.is_empty() { + return ( + StatusCode::BAD_REQUEST, + Json(json!({"error": "InvalidRequest", "message": "No CIDs provided"})), + ) + .into_response(); + } + let blocks_res = state.block_store.get_many(&cids).await; let blocks = match blocks_res { Ok(blocks) => blocks, Err(e) => { error!("Failed to get blocks: {}", e); - return (StatusCode::INTERNAL_SERVER_ERROR, "Failed to get blocks").into_response(); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({"error": "InternalError", "message": "Failed to get blocks"})), + ) + .into_response(); } }; - if cids.is_empty() { - return (StatusCode::BAD_REQUEST, "No CIDs provided").into_response(); + + let mut missing_cids: Vec = Vec::new(); + for (i, block_opt) in blocks.iter().enumerate() { + if block_opt.is_none() { + missing_cids.push(cids[i].to_string()); + } } - let root_cid = cids[0]; - let header = match encode_car_header(&root_cid) { + if !missing_cids.is_empty() { + return ( + StatusCode::BAD_REQUEST, + Json(json!({ + "error": "InvalidRequest", + "message": format!("Could not find blocks: {}", missing_cids.join(", ")) + })), + ) + .into_response(); + } + + let header = match crate::sync::car::encode_car_header_null_root() { Ok(h) => h, Err(e) => { error!("Failed to encode CAR header: {}", e); - return (StatusCode::INTERNAL_SERVER_ERROR, "Failed to encode CAR").into_response(); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({"error": "InternalError", "message": "Failed to encode CAR"})), + ) + .into_response(); } }; let mut car_bytes = header; @@ -97,40 +152,22 @@ pub async fn get_repo( State(state): State, Query(query): Query, ) -> Response { - let repo_row = sqlx::query!( - r#" - SELECT r.repo_root_cid - FROM repos r - JOIN users u ON u.id = r.user_id - WHERE u.did = $1 - "#, - query.did - ) - .fetch_optional(&state.db) - .await - .unwrap_or(None); - let head_str = match repo_row { - Some(r) => r.repo_root_cid, + let account = match assert_repo_availability(&state.db, &query.did, false).await { + Ok(a) => a, + Err(e) => return e.into_response(), + }; + + let head_str = match account.repo_root_cid { + Some(cid) => cid, None => { - let user_exists = sqlx::query!("SELECT id FROM users WHERE did = $1", query.did) - .fetch_optional(&state.db) - .await - .unwrap_or(None); - if user_exists.is_none() { - return ( - StatusCode::NOT_FOUND, - Json(json!({"error": "RepoNotFound", "message": "Repo not found"})), - ) - .into_response(); - } else { - return ( - StatusCode::NOT_FOUND, - Json(json!({"error": "RepoNotFound", "message": "Repo not initialized"})), - ) - .into_response(); - } + return ( + StatusCode::BAD_REQUEST, + Json(json!({"error": "RepoNotFound", "message": "Repo not initialized"})), + ) + .into_response(); } }; + let head_cid = match Cid::from_str(&head_str) { Ok(c) => c, Err(_) => { @@ -141,6 +178,11 @@ pub async fn get_repo( .into_response(); } }; + + if let Some(since) = &query.since { + return get_repo_since(&state, &query.did, &head_cid, since).await; + } + let mut car_bytes = match encode_car_header(&head_cid) { Ok(h) => h, Err(e) => { @@ -189,6 +231,109 @@ pub async fn get_repo( .into_response() } +async fn get_repo_since(state: &AppState, did: &str, head_cid: &Cid, since: &str) -> Response { + let events = sqlx::query!( + r#" + SELECT blocks_cids, commit_cid + FROM repo_seq + WHERE did = $1 AND rev > $2 + ORDER BY seq DESC + "#, + did, + since + ) + .fetch_all(&state.db) + .await; + + let events = match events { + Ok(e) => e, + Err(e) => { + error!("DB error in get_repo_since: {:?}", e); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({"error": "InternalError", "message": "Database error"})), + ) + .into_response(); + } + }; + + let mut block_cids: Vec = Vec::new(); + for event in &events { + if let Some(cids) = &event.blocks_cids { + for cid_str in cids { + if let Ok(cid) = Cid::from_str(cid_str) + && !block_cids.contains(&cid) + { + block_cids.push(cid); + } + } + } + if let Some(commit_cid_str) = &event.commit_cid + && let Ok(cid) = Cid::from_str(commit_cid_str) + && !block_cids.contains(&cid) + { + block_cids.push(cid); + } + } + + let mut car_bytes = match encode_car_header(head_cid) { + Ok(h) => h, + Err(e) => { + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({"error": "InternalError", "message": format!("Failed to encode CAR header: {}", e)})), + ) + .into_response(); + } + }; + + if block_cids.is_empty() { + return ( + StatusCode::OK, + [(axum::http::header::CONTENT_TYPE, "application/vnd.ipld.car")], + car_bytes, + ) + .into_response(); + } + + let blocks = match state.block_store.get_many(&block_cids).await { + Ok(b) => b, + Err(e) => { + error!("Block store error in get_repo_since: {:?}", e); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({"error": "InternalError", "message": "Failed to get blocks"})), + ) + .into_response(); + } + }; + + for (i, block_opt) in blocks.into_iter().enumerate() { + if let Some(block) = block_opt { + let cid = block_cids[i]; + let cid_bytes = cid.to_bytes(); + let total_len = cid_bytes.len() + block.len(); + let mut writer = Vec::new(); + crate::sync::car::write_varint(&mut writer, total_len as u64) + .expect("Writing to Vec should never fail"); + writer + .write_all(&cid_bytes) + .expect("Writing to Vec should never fail"); + writer + .write_all(&block) + .expect("Writing to Vec should never fail"); + car_bytes.extend_from_slice(&writer); + } + } + + ( + StatusCode::OK, + [(axum::http::header::CONTENT_TYPE, "application/vnd.ipld.car")], + car_bytes, + ) + .into_response() +} + fn extract_links_ipld(value: &Ipld, stack: &mut Vec) { match value { Ipld::Link(cid) => { @@ -224,24 +369,17 @@ pub async fn get_record( use std::collections::BTreeMap; use std::sync::Arc; - let repo_row = sqlx::query!( - r#" - SELECT r.repo_root_cid - FROM repos r - JOIN users u ON u.id = r.user_id - WHERE u.did = $1 - "#, - query.did - ) - .fetch_optional(&state.db) - .await - .unwrap_or(None); - let commit_cid_str = match repo_row { - Some(r) => r.repo_root_cid, + let account = match assert_repo_availability(&state.db, &query.did, false).await { + Ok(a) => a, + Err(e) => return e.into_response(), + }; + + let commit_cid_str = match account.repo_root_cid { + Some(cid) => cid, None => { return ( - StatusCode::NOT_FOUND, - Json(json!({"error": "RepoNotFound", "message": "Repo not found"})), + StatusCode::BAD_REQUEST, + Json(json!({"error": "RepoNotFound", "message": "Repo not initialized"})), ) .into_response(); } diff --git a/src/sync/subscribe_repos.rs b/src/sync/subscribe_repos.rs index 612c312..0a75a36 100644 --- a/src/sync/subscribe_repos.rs +++ b/src/sync/subscribe_repos.rs @@ -1,7 +1,8 @@ use crate::state::AppState; use crate::sync::firehose::SequencedEvent; use crate::sync::util::{ - format_event_for_sending, format_event_with_prefetched_blocks, prefetch_blocks_for_events, + format_error_frame, format_event_for_sending, format_event_with_prefetched_blocks, + format_info_frame, prefetch_blocks_for_events, }; use axum::{ extract::{Query, State, ws::Message, ws::WebSocket, ws::WebSocketUpgrade}, @@ -55,13 +56,85 @@ async fn handle_socket(mut socket: WebSocket, state: AppState, params: Subscribe info!(subscribers = count, "Firehose subscriber disconnected"); } +fn get_backfill_hours() -> i64 { + std::env::var("FIREHOSE_BACKFILL_HOURS") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(72) +} + async fn handle_socket_inner( socket: &mut WebSocket, state: &AppState, params: SubscribeReposParams, ) -> Result<(), ()> { + let mut rx = state.firehose_tx.subscribe(); + let mut last_seen: i64 = -1; + if let Some(cursor) = params.cursor { + let current_seq = sqlx::query_scalar!("SELECT MAX(seq) FROM repo_seq") + .fetch_one(&state.db) + .await + .ok() + .flatten() + .unwrap_or(0); + + if cursor > current_seq { + if let Ok(error_bytes) = + format_error_frame("FutureCursor", Some("Cursor in the future.")) + { + let _ = socket.send(Message::Binary(error_bytes.into())).await; + } + socket.close().await.ok(); + return Err(()); + } + + let backfill_time = chrono::Utc::now() - chrono::Duration::hours(get_backfill_hours()); + + let first_event = sqlx::query_as!( + SequencedEvent, + r#" + SELECT seq, did, created_at, event_type, commit_cid, prev_cid, prev_data_cid, ops, blobs, blocks_cids, handle, active, status, rev + FROM repo_seq + WHERE seq > $1 + ORDER BY seq ASC + LIMIT 1 + "#, + cursor + ) + .fetch_optional(&state.db) + .await + .ok() + .flatten(); + let mut current_cursor = cursor; + + if let Some(ref event) = first_event + && event.created_at < backfill_time + { + if let Ok(info_bytes) = format_info_frame( + "OutdatedCursor", + Some("Requested cursor exceeded limit. Possibly missing events"), + ) { + let _ = socket.send(Message::Binary(info_bytes.into())).await; + } + + let earliest = sqlx::query_scalar!( + "SELECT MIN(seq) FROM repo_seq WHERE created_at >= $1", + backfill_time + ) + .fetch_one(&state.db) + .await + .ok() + .flatten(); + + if let Some(earliest_seq) = earliest { + current_cursor = earliest_seq - 1; + } + } + + last_seen = current_cursor; + loop { let events = sqlx::query_as!( SequencedEvent, @@ -93,6 +166,7 @@ async fn handle_socket_inner( }; for event in events { current_cursor = event.seq; + last_seen = event.seq; let bytes = match format_event_with_prefetched_blocks(event, &prefetched).await { Ok(b) => b, @@ -118,8 +192,48 @@ async fn handle_socket_inner( } } } + + let cutover_events = sqlx::query_as!( + SequencedEvent, + r#" + SELECT seq, did, created_at, event_type, commit_cid, prev_cid, prev_data_cid, ops, blobs, blocks_cids, handle, active, status, rev + FROM repo_seq + WHERE seq > $1 + ORDER BY seq ASC + "#, + last_seen + ) + .fetch_all(&state.db) + .await; + + if let Ok(events) = cutover_events + && !events.is_empty() + { + let prefetched = match prefetch_blocks_for_events(state, &events).await { + Ok(blocks) => blocks, + Err(e) => { + error!("Failed to prefetch blocks for cutover: {}", e); + socket.close().await.ok(); + return Err(()); + } + }; + for event in events { + last_seen = event.seq; + let bytes = match format_event_with_prefetched_blocks(event, &prefetched).await { + Ok(b) => b, + Err(e) => { + warn!("Failed to format cutover event: {}", e); + return Err(()); + } + }; + if let Err(e) = socket.send(Message::Binary(bytes.into())).await { + warn!("Failed to send cutover event: {}", e); + return Err(()); + } + crate::metrics::record_firehose_event(); + } + } } - let mut rx = state.firehose_tx.subscribe(); let max_lag_before_disconnect: u64 = std::env::var("FIREHOSE_MAX_LAG") .ok() .and_then(|v| v.parse().ok()) @@ -129,6 +243,10 @@ async fn handle_socket_inner( result = rx.recv() => { match result { Ok(event) => { + if event.seq <= last_seen { + continue; + } + last_seen = event.seq; if let Err(e) = send_event(socket, state, event).await { warn!("Failed to send event: {}", e); break; diff --git a/src/sync/util.rs b/src/sync/util.rs index e8c6bde..d7d2295 100644 --- a/src/sync/util.rs +++ b/src/sync/util.rs @@ -1,16 +1,167 @@ use crate::state::AppState; use crate::sync::firehose::SequencedEvent; -use crate::sync::frame::{AccountFrame, CommitFrame, FrameHeader, IdentityFrame, SyncFrame}; +use crate::sync::frame::{ + AccountFrame, CommitFrame, ErrorFrameBody, ErrorFrameHeader, FrameHeader, IdentityFrame, + InfoFrame, SyncFrame, +}; +use axum::Json; +use axum::http::StatusCode; +use axum::response::{IntoResponse, Response}; use bytes::Bytes; use cid::Cid; use iroh_car::{CarHeader, CarWriter}; use jacquard_repo::commit::Commit; use jacquard_repo::storage::BlockStore; +use serde::Serialize; +use serde_json::json; +use sqlx::PgPool; use std::collections::{BTreeMap, HashMap}; use std::io::Cursor; use std::str::FromStr; use tokio::io::AsyncWriteExt; +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[serde(rename_all = "lowercase")] +pub enum AccountStatus { + Active, + Takendown, + Suspended, + Deactivated, + Deleted, +} + +impl AccountStatus { + pub fn as_str(&self) -> Option<&'static str> { + match self { + AccountStatus::Active => None, + AccountStatus::Takendown => Some("takendown"), + AccountStatus::Suspended => Some("suspended"), + AccountStatus::Deactivated => Some("deactivated"), + AccountStatus::Deleted => Some("deleted"), + } + } + + pub fn is_active(&self) -> bool { + matches!(self, AccountStatus::Active) + } +} + +pub struct RepoAccount { + pub did: String, + pub user_id: uuid::Uuid, + pub status: AccountStatus, + pub repo_root_cid: Option, +} + +pub enum RepoAvailabilityError { + NotFound(String), + Takendown(String), + Deactivated(String), + Internal(String), +} + +impl IntoResponse for RepoAvailabilityError { + fn into_response(self) -> Response { + match self { + RepoAvailabilityError::NotFound(did) => ( + StatusCode::BAD_REQUEST, + Json(json!({ + "error": "RepoNotFound", + "message": format!("Could not find repo for DID: {}", did) + })), + ) + .into_response(), + RepoAvailabilityError::Takendown(did) => ( + StatusCode::BAD_REQUEST, + Json(json!({ + "error": "RepoTakendown", + "message": format!("Repo has been takendown: {}", did) + })), + ) + .into_response(), + RepoAvailabilityError::Deactivated(did) => ( + StatusCode::BAD_REQUEST, + Json(json!({ + "error": "RepoDeactivated", + "message": format!("Repo has been deactivated: {}", did) + })), + ) + .into_response(), + RepoAvailabilityError::Internal(msg) => ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({ + "error": "InternalError", + "message": msg + })), + ) + .into_response(), + } + } +} + +pub async fn get_account_with_status( + db: &PgPool, + did: &str, +) -> Result, sqlx::Error> { + let row = sqlx::query!( + r#" + SELECT u.id, u.did, u.deactivated_at, u.takedown_ref, r.repo_root_cid + FROM users u + LEFT JOIN repos r ON r.user_id = u.id + WHERE u.did = $1 + "#, + did + ) + .fetch_optional(db) + .await?; + + Ok(row.map(|r| { + let status = if r.takedown_ref.is_some() { + AccountStatus::Takendown + } else if r.deactivated_at.is_some() { + AccountStatus::Deactivated + } else { + AccountStatus::Active + }; + + RepoAccount { + did: r.did, + user_id: r.id, + status, + repo_root_cid: Some(r.repo_root_cid), + } + })) +} + +pub async fn assert_repo_availability( + db: &PgPool, + did: &str, + is_admin_or_self: bool, +) -> Result { + let account = get_account_with_status(db, did) + .await + .map_err(|e| RepoAvailabilityError::Internal(e.to_string()))?; + + let account = match account { + Some(a) => a, + None => return Err(RepoAvailabilityError::NotFound(did.to_string())), + }; + + if is_admin_or_self { + return Ok(account); + } + + match account.status { + AccountStatus::Takendown => return Err(RepoAvailabilityError::Takendown(did.to_string())), + AccountStatus::Deactivated => { + return Err(RepoAvailabilityError::Deactivated(did.to_string())); + } + _ => {} + } + + Ok(account) +} + fn extract_rev_from_commit_bytes(commit_bytes: &[u8]) -> Option { Commit::from_cbor(commit_bytes) .ok() @@ -351,3 +502,30 @@ pub async fn format_event_with_prefetched_blocks( serde_ipld_dagcbor::to_writer(&mut bytes, &frame)?; Ok(bytes) } + +pub fn format_info_frame(name: &str, message: Option<&str>) -> Result, anyhow::Error> { + let header = FrameHeader { + op: 1, + t: "#info".to_string(), + }; + let frame = InfoFrame { + name: name.to_string(), + message: message.map(String::from), + }; + let mut bytes = Vec::new(); + serde_ipld_dagcbor::to_writer(&mut bytes, &header)?; + serde_ipld_dagcbor::to_writer(&mut bytes, &frame)?; + Ok(bytes) +} + +pub fn format_error_frame(error: &str, message: Option<&str>) -> Result, anyhow::Error> { + let header = ErrorFrameHeader { op: -1 }; + let frame = ErrorFrameBody { + error: error.to_string(), + message: message.map(String::from), + }; + let mut bytes = Vec::new(); + serde_ipld_dagcbor::to_writer(&mut bytes, &header)?; + serde_ipld_dagcbor::to_writer(&mut bytes, &frame)?; + Ok(bytes) +} diff --git a/src/util.rs b/src/util.rs index bf173b9..9b009ca 100644 --- a/src/util.rs +++ b/src/util.rs @@ -73,6 +73,33 @@ pub async fn get_user_by_identifier( .ok_or(DbLookupError::NotFound) } +pub fn parse_repeated_query_param(query: Option<&str>, key: &str) -> Vec { + query + .map(|q| { + let mut values = Vec::new(); + for pair in q.split('&') { + if let Some((k, v)) = pair.split_once('=') + && k == key + && let Ok(decoded) = urlencoding::decode(v) + { + let decoded = decoded.into_owned(); + if decoded.contains(',') { + for part in decoded.split(',') { + let trimmed = part.trim(); + if !trimmed.is_empty() { + values.push(trimmed.to_string()); + } + } + } else if !decoded.is_empty() { + values.push(decoded); + } + } + } + values + }) + .unwrap_or_default() +} + pub fn extract_client_ip(headers: &HeaderMap) -> String { if let Some(forwarded) = headers.get("x-forwarded-for") && let Ok(value) = forwarded.to_str() @@ -92,6 +119,48 @@ pub fn extract_client_ip(headers: &HeaderMap) -> String { mod tests { use super::*; + #[test] + fn test_parse_repeated_query_param_repeated() { + let query = "did=test&cids=a&cids=b&cids=c"; + let result = parse_repeated_query_param(Some(query), "cids"); + assert_eq!(result, vec!["a", "b", "c"]); + } + + #[test] + fn test_parse_repeated_query_param_comma_separated() { + let query = "did=test&cids=a,b,c"; + let result = parse_repeated_query_param(Some(query), "cids"); + assert_eq!(result, vec!["a", "b", "c"]); + } + + #[test] + fn test_parse_repeated_query_param_mixed() { + let query = "did=test&cids=a,b&cids=c"; + let result = parse_repeated_query_param(Some(query), "cids"); + assert_eq!(result, vec!["a", "b", "c"]); + } + + #[test] + fn test_parse_repeated_query_param_single() { + let query = "did=test&cids=a"; + let result = parse_repeated_query_param(Some(query), "cids"); + assert_eq!(result, vec!["a"]); + } + + #[test] + fn test_parse_repeated_query_param_empty() { + let query = "did=test"; + let result = parse_repeated_query_param(Some(query), "cids"); + assert!(result.is_empty()); + } + + #[test] + fn test_parse_repeated_query_param_url_encoded() { + let query = "did=test&cids=bafyreib%2Btest"; + let result = parse_repeated_query_param(Some(query), "cids"); + assert_eq!(result, vec!["bafyreib+test"]); + } + #[test] fn test_generate_token_code() { let code = generate_token_code(); diff --git a/tests/firehose_validation.rs b/tests/firehose_validation.rs index 7cb7214..3e69705 100644 --- a/tests/firehose_validation.rs +++ b/tests/firehose_validation.rs @@ -200,6 +200,7 @@ async fn test_firehose_frame_structure() { app_port() ); let (mut ws_stream, _) = connect_async(&url).await.expect("Failed to connect"); + tokio::time::sleep(std::time::Duration::from_millis(100)).await; let post_text = "Testing firehose validation!"; let post_payload = json!({ @@ -224,7 +225,7 @@ async fn test_firehose_frame_structure() { assert_eq!(res.status(), StatusCode::OK); let mut frame_opt: Option<(FrameHeader, CommitFrame)> = None; - let timeout = tokio::time::timeout(std::time::Duration::from_secs(5), async { + let timeout = tokio::time::timeout(std::time::Duration::from_secs(10), async { loop { let msg = ws_stream.next().await.unwrap().unwrap(); let raw_bytes = match msg { @@ -392,6 +393,7 @@ async fn test_firehose_update_has_prev_field() { app_port() ); let (mut ws_stream, _) = connect_async(&url).await.expect("Failed to connect"); + tokio::time::sleep(std::time::Duration::from_millis(100)).await; let update_payload = json!({ "repo": did, @@ -415,7 +417,7 @@ async fn test_firehose_update_has_prev_field() { assert_eq!(res.status(), StatusCode::OK); let mut frame_opt: Option = None; - let timeout = tokio::time::timeout(std::time::Duration::from_secs(15), async { + let timeout = tokio::time::timeout(std::time::Duration::from_secs(20), async { loop { let msg = match ws_stream.next().await { Some(Ok(m)) => m, @@ -472,6 +474,7 @@ async fn test_firehose_commit_has_prev_data() { app_port() ); let (mut ws_stream, _) = connect_async(&url).await.expect("Failed to connect"); + tokio::time::sleep(std::time::Duration::from_millis(100)).await; let post_payload = json!({ "repo": did, @@ -494,7 +497,7 @@ async fn test_firehose_commit_has_prev_data() { .expect("Failed to create first post"); let mut first_frame_opt: Option = None; - let timeout = tokio::time::timeout(std::time::Duration::from_secs(5), async { + let timeout = tokio::time::timeout(std::time::Duration::from_secs(10), async { loop { let msg = ws_stream.next().await.unwrap().unwrap(); let raw_bytes = match msg { @@ -544,7 +547,7 @@ async fn test_firehose_commit_has_prev_data() { .expect("Failed to create second post"); let mut second_frame_opt: Option = None; - let timeout = tokio::time::timeout(std::time::Duration::from_secs(5), async { + let timeout = tokio::time::timeout(std::time::Duration::from_secs(10), async { loop { let msg = ws_stream.next().await.unwrap().unwrap(); let raw_bytes = match msg { @@ -593,6 +596,7 @@ async fn test_compare_raw_cbor_encoding() { app_port() ); let (mut ws_stream, _) = connect_async(&url).await.expect("Failed to connect"); + tokio::time::sleep(std::time::Duration::from_millis(100)).await; let post_payload = json!({ "repo": did, @@ -615,7 +619,7 @@ async fn test_compare_raw_cbor_encoding() { .expect("Failed to create post"); let mut raw_bytes_opt: Option> = None; - let timeout = tokio::time::timeout(std::time::Duration::from_secs(5), async { + let timeout = tokio::time::timeout(std::time::Duration::from_secs(10), async { loop { let msg = ws_stream.next().await.unwrap().unwrap(); let raw = match msg { @@ -661,3 +665,188 @@ async fn test_compare_raw_cbor_encoding() { ws_stream.send(tungstenite::Message::Close(None)).await.ok(); } + +#[derive(Debug, Deserialize)] +struct ErrorFrameHeader { + op: i64, +} + +#[derive(Debug, Deserialize)] +struct ErrorFrameBody { + error: String, + #[allow(dead_code)] + message: Option, +} + +#[derive(Debug, Deserialize)] +struct InfoFrameHeader { + #[allow(dead_code)] + op: i64, + t: String, +} + +#[derive(Debug, Deserialize)] +struct InfoFrameBody { + name: String, + #[allow(dead_code)] + message: Option, +} + +fn parse_error_frame(bytes: &[u8]) -> Result<(ErrorFrameHeader, ErrorFrameBody), String> { + let header_len = find_cbor_map_end(bytes)?; + let header: ErrorFrameHeader = serde_ipld_dagcbor::from_slice(&bytes[..header_len]) + .map_err(|e| format!("Failed to parse error header: {:?}", e))?; + + if header.op != -1 { + return Err(format!("Not an error frame, op: {}", header.op)); + } + + let remaining = &bytes[header_len..]; + let body: ErrorFrameBody = serde_ipld_dagcbor::from_slice(remaining) + .map_err(|e| format!("Failed to parse error body: {:?}", e))?; + + Ok((header, body)) +} + +fn parse_info_frame(bytes: &[u8]) -> Result<(InfoFrameHeader, InfoFrameBody), String> { + let header_len = find_cbor_map_end(bytes)?; + let header: InfoFrameHeader = serde_ipld_dagcbor::from_slice(&bytes[..header_len]) + .map_err(|e| format!("Failed to parse info header: {:?}", e))?; + + if header.t != "#info" { + return Err(format!("Not an info frame, t: {}", header.t)); + } + + let remaining = &bytes[header_len..]; + let body: InfoFrameBody = serde_ipld_dagcbor::from_slice(remaining) + .map_err(|e| format!("Failed to parse info body: {:?}", e))?; + + Ok((header, body)) +} + +#[tokio::test] +async fn test_firehose_future_cursor_error() { + let _ = base_url().await; + + let future_cursor = 9999999999i64; + let url = format!( + "ws://127.0.0.1:{}/xrpc/com.atproto.sync.subscribeRepos?cursor={}", + app_port(), + future_cursor + ); + + let (mut ws_stream, _) = connect_async(&url).await.expect("Failed to connect"); + + let timeout = tokio::time::timeout(std::time::Duration::from_secs(10), async { + loop { + match ws_stream.next().await { + Some(Ok(tungstenite::Message::Binary(bin))) => { + if let Ok((header, body)) = parse_error_frame(&bin) { + println!("Received error frame: {:?} {:?}", header, body); + assert_eq!(header.op, -1, "Error frame op should be -1"); + assert_eq!(body.error, "FutureCursor", "Error should be FutureCursor"); + return true; + } + } + Some(Ok(tungstenite::Message::Close(_))) => { + println!("Connection closed"); + return false; + } + None => { + println!("Stream ended"); + return false; + } + _ => continue, + } + } + }) + .await; + + match timeout { + Ok(received_error) => { + assert!( + received_error, + "Should have received FutureCursor error frame before connection closed" + ); + } + Err(_) => { + panic!( + "Timed out waiting for FutureCursor error - connection should close quickly with error" + ); + } + } +} + +#[tokio::test] +async fn test_firehose_outdated_cursor_info() { + let client = client(); + let (token, did) = create_account_and_login(&client).await; + + let post_payload = json!({ + "repo": did, + "collection": "app.bsky.feed.post", + "record": { + "$type": "app.bsky.feed.post", + "text": "Post for outdated cursor test", + "createdAt": chrono::Utc::now().to_rfc3339(), + } + }); + let _ = client + .post(format!( + "{}/xrpc/com.atproto.repo.createRecord", + base_url().await + )) + .bearer_auth(&token) + .json(&post_payload) + .send() + .await + .expect("Failed to create post"); + + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + + let outdated_cursor = 1i64; + let url = format!( + "ws://127.0.0.1:{}/xrpc/com.atproto.sync.subscribeRepos?cursor={}", + app_port(), + outdated_cursor + ); + + let (mut ws_stream, _) = connect_async(&url).await.expect("Failed to connect"); + + let mut found_info = false; + let mut found_commit = false; + + let timeout = tokio::time::timeout(std::time::Duration::from_secs(15), async { + loop { + match ws_stream.next().await { + Some(Ok(tungstenite::Message::Binary(bin))) => { + if let Ok((header, body)) = parse_info_frame(&bin) { + println!("Received info frame: {:?} {:?}", header, body); + if body.name == "OutdatedCursor" { + found_info = true; + println!("Found OutdatedCursor info frame!"); + } + } else if let Ok((_, frame)) = parse_frame(&bin) { + if frame.repo == did { + found_commit = true; + println!("Found commit for our DID"); + } + } + if found_commit { + break; + } + } + Some(Ok(tungstenite::Message::Close(_))) => break, + None => break, + _ => continue, + } + } + }) + .await; + + assert!(timeout.is_ok(), "Timed out"); + assert!( + found_commit, + "Should have received commits even with outdated cursor" + ); +} diff --git a/tests/helpers/mod.rs b/tests/helpers/mod.rs index 65ba9bd..489145b 100644 --- a/tests/helpers/mod.rs +++ b/tests/helpers/mod.rs @@ -214,3 +214,41 @@ pub async fn create_repost( body["cid"].as_str().unwrap().to_string(), ) } + +#[allow(dead_code)] +pub async fn set_account_takedown(did: &str, takedown_ref: Option<&str>) { + let conn_str = get_db_connection_string().await; + let pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(2) + .connect(&conn_str) + .await + .expect("Failed to connect to test database"); + sqlx::query!( + "UPDATE users SET takedown_ref = $1 WHERE did = $2", + takedown_ref, + did + ) + .execute(&pool) + .await + .expect("Failed to update takedown_ref"); +} + +#[allow(dead_code)] +pub async fn set_account_deactivated(did: &str, deactivated: bool) { + let conn_str = get_db_connection_string().await; + let pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(2) + .connect(&conn_str) + .await + .expect("Failed to connect to test database"); + let deactivated_at: Option> = + if deactivated { Some(Utc::now()) } else { None }; + sqlx::query!( + "UPDATE users SET deactivated_at = $1 WHERE did = $2", + deactivated_at, + did + ) + .execute(&pool) + .await + .expect("Failed to update deactivated_at"); +} diff --git a/tests/sync_blob.rs b/tests/sync_blob.rs index d6aca99..7ac6168 100644 --- a/tests/sync_blob.rs +++ b/tests/sync_blob.rs @@ -50,7 +50,7 @@ async fn test_list_blobs_not_found() { .send() .await .expect("Failed to send request"); - assert_eq!(res.status(), StatusCode::NOT_FOUND); + assert_eq!(res.status(), StatusCode::BAD_REQUEST); let body: Value = res.json().await.expect("Response was not valid JSON"); assert_eq!(body["error"], "RepoNotFound"); } diff --git a/tests/sync_conformance.rs b/tests/sync_conformance.rs new file mode 100644 index 0000000..0ba5590 --- /dev/null +++ b/tests/sync_conformance.rs @@ -0,0 +1,443 @@ +mod common; +mod helpers; + +use common::*; +use helpers::*; +use reqwest::StatusCode; +use serde_json::Value; + +#[tokio::test] +async fn test_get_repo_takendown_returns_error() { + let client = client(); + let (_, did) = create_account_and_login(&client).await; + + set_account_takedown(&did, Some("test-takedown-ref")).await; + + let res = client + .get(format!( + "{}/xrpc/com.atproto.sync.getRepo", + base_url().await + )) + .query(&[("did", did.as_str())]) + .send() + .await + .expect("Failed to send request"); + + assert_eq!(res.status(), StatusCode::BAD_REQUEST); + let body: Value = res.json().await.expect("Response was not valid JSON"); + assert_eq!(body["error"], "RepoTakendown"); +} + +#[tokio::test] +async fn test_get_repo_deactivated_returns_error() { + let client = client(); + let (_, did) = create_account_and_login(&client).await; + + set_account_deactivated(&did, true).await; + + let res = client + .get(format!( + "{}/xrpc/com.atproto.sync.getRepo", + base_url().await + )) + .query(&[("did", did.as_str())]) + .send() + .await + .expect("Failed to send request"); + + assert_eq!(res.status(), StatusCode::BAD_REQUEST); + let body: Value = res.json().await.expect("Response was not valid JSON"); + assert_eq!(body["error"], "RepoDeactivated"); +} + +#[tokio::test] +async fn test_get_latest_commit_takendown_returns_error() { + let client = client(); + let (_, did) = create_account_and_login(&client).await; + + set_account_takedown(&did, Some("test-takedown-ref")).await; + + let res = client + .get(format!( + "{}/xrpc/com.atproto.sync.getLatestCommit", + base_url().await + )) + .query(&[("did", did.as_str())]) + .send() + .await + .expect("Failed to send request"); + + assert_eq!(res.status(), StatusCode::BAD_REQUEST); + let body: Value = res.json().await.expect("Response was not valid JSON"); + assert_eq!(body["error"], "RepoTakendown"); +} + +#[tokio::test] +async fn test_get_blocks_takendown_returns_error() { + let client = client(); + let (_, did) = create_account_and_login(&client).await; + + let commit_res = client + .get(format!( + "{}/xrpc/com.atproto.sync.getLatestCommit", + base_url().await + )) + .query(&[("did", did.as_str())]) + .send() + .await + .expect("Failed to get commit"); + let commit_body: Value = commit_res.json().await.unwrap(); + let cid = commit_body["cid"].as_str().unwrap(); + + set_account_takedown(&did, Some("test-takedown-ref")).await; + + let res = client + .get(format!( + "{}/xrpc/com.atproto.sync.getBlocks", + base_url().await + )) + .query(&[("did", did.as_str()), ("cids", cid)]) + .send() + .await + .expect("Failed to send request"); + + assert_eq!(res.status(), StatusCode::BAD_REQUEST); + let body: Value = res.json().await.expect("Response was not valid JSON"); + assert_eq!(body["error"], "RepoTakendown"); +} + +#[tokio::test] +async fn test_get_repo_status_shows_takendown_status() { + let client = client(); + let (_, did) = create_account_and_login(&client).await; + + set_account_takedown(&did, Some("test-takedown-ref")).await; + + let res = client + .get(format!( + "{}/xrpc/com.atproto.sync.getRepoStatus", + base_url().await + )) + .query(&[("did", did.as_str())]) + .send() + .await + .expect("Failed to send request"); + + assert_eq!(res.status(), StatusCode::OK); + let body: Value = res.json().await.expect("Response was not valid JSON"); + assert_eq!(body["active"], false); + assert_eq!(body["status"], "takendown"); + assert!(body.get("rev").is_none() || body["rev"].is_null()); +} + +#[tokio::test] +async fn test_get_repo_status_shows_deactivated_status() { + let client = client(); + let (_, did) = create_account_and_login(&client).await; + + set_account_deactivated(&did, true).await; + + let res = client + .get(format!( + "{}/xrpc/com.atproto.sync.getRepoStatus", + base_url().await + )) + .query(&[("did", did.as_str())]) + .send() + .await + .expect("Failed to send request"); + + assert_eq!(res.status(), StatusCode::OK); + let body: Value = res.json().await.expect("Response was not valid JSON"); + assert_eq!(body["active"], false); + assert_eq!(body["status"], "deactivated"); +} + +#[tokio::test] +async fn test_list_repos_shows_status_field() { + let client = client(); + let (_, did) = create_account_and_login(&client).await; + + set_account_takedown(&did, Some("test-takedown-ref")).await; + + let res = client + .get(format!( + "{}/xrpc/com.atproto.sync.listRepos", + base_url().await + )) + .send() + .await + .expect("Failed to send request"); + + assert_eq!(res.status(), StatusCode::OK); + let body: Value = res.json().await.expect("Response was not valid JSON"); + let repos = body["repos"].as_array().unwrap(); + + let takendown_repo = repos.iter().find(|r| r["did"] == did); + assert!(takendown_repo.is_some(), "Takendown repo should be in list"); + let repo = takendown_repo.unwrap(); + assert_eq!(repo["active"], false); + assert_eq!(repo["status"], "takendown"); +} + +#[tokio::test] +async fn test_get_blob_takendown_returns_error() { + let client = client(); + let (jwt, did) = create_account_and_login(&client).await; + + let blob_res = client + .post(format!( + "{}/xrpc/com.atproto.repo.uploadBlob", + base_url().await + )) + .header("Content-Type", "image/png") + .bearer_auth(&jwt) + .body(vec![0x89, 0x50, 0x4E, 0x47, 0x0D, 0x0A, 0x1A, 0x0A]) + .send() + .await + .expect("Failed to upload blob"); + let blob_body: Value = blob_res.json().await.unwrap(); + let cid = blob_body["blob"]["ref"]["$link"].as_str().unwrap(); + + set_account_takedown(&did, Some("test-takedown-ref")).await; + + let res = client + .get(format!( + "{}/xrpc/com.atproto.sync.getBlob", + base_url().await + )) + .query(&[("did", did.as_str()), ("cid", cid)]) + .send() + .await + .expect("Failed to send request"); + + assert_eq!(res.status(), StatusCode::BAD_REQUEST); + let body: Value = res.json().await.expect("Response was not valid JSON"); + assert_eq!(body["error"], "RepoTakendown"); +} + +#[tokio::test] +async fn test_get_blob_has_security_headers() { + let client = client(); + let (jwt, did) = create_account_and_login(&client).await; + + let blob_res = client + .post(format!( + "{}/xrpc/com.atproto.repo.uploadBlob", + base_url().await + )) + .header("Content-Type", "image/png") + .bearer_auth(&jwt) + .body(vec![0x89, 0x50, 0x4E, 0x47, 0x0D, 0x0A, 0x1A, 0x0A]) + .send() + .await + .expect("Failed to upload blob"); + let blob_body: Value = blob_res.json().await.unwrap(); + let cid = blob_body["blob"]["ref"]["$link"].as_str().unwrap(); + + let res = client + .get(format!( + "{}/xrpc/com.atproto.sync.getBlob", + base_url().await + )) + .query(&[("did", did.as_str()), ("cid", cid)]) + .send() + .await + .expect("Failed to send request"); + + assert_eq!(res.status(), StatusCode::OK); + + let headers = res.headers(); + assert_eq!( + headers + .get("x-content-type-options") + .map(|v| v.to_str().unwrap()), + Some("nosniff"), + "Missing x-content-type-options: nosniff header" + ); + assert_eq!( + headers + .get("content-security-policy") + .map(|v| v.to_str().unwrap()), + Some("default-src 'none'; sandbox"), + "Missing content-security-policy header" + ); + assert!( + headers.get("content-length").is_some(), + "Missing content-length header" + ); +} + +#[tokio::test] +async fn test_get_blocks_missing_cids_returns_error() { + let client = client(); + let (_, did) = create_account_and_login(&client).await; + + let fake_cid = "bafyreif2pall7dybz7vecqka3zo24irdwabwdi4wc55jznaq75q7eaavvu"; + + let res = client + .get(format!( + "{}/xrpc/com.atproto.sync.getBlocks", + base_url().await + )) + .query(&[("did", did.as_str()), ("cids", fake_cid)]) + .send() + .await + .expect("Failed to send request"); + + assert_eq!(res.status(), StatusCode::BAD_REQUEST); + let body: Value = res.json().await.expect("Response was not valid JSON"); + assert_eq!(body["error"], "InvalidRequest"); + assert!( + body["message"] + .as_str() + .unwrap() + .contains("Could not find blocks"), + "Error message should mention missing blocks" + ); +} + +#[tokio::test] +async fn test_get_blocks_accepts_array_format() { + let client = client(); + let (_, did) = create_account_and_login(&client).await; + + let commit_res = client + .get(format!( + "{}/xrpc/com.atproto.sync.getLatestCommit", + base_url().await + )) + .query(&[("did", did.as_str())]) + .send() + .await + .expect("Failed to get commit"); + let commit_body: Value = commit_res.json().await.unwrap(); + let cid = commit_body["cid"].as_str().unwrap(); + + let url = format!( + "{}/xrpc/com.atproto.sync.getBlocks?did={}&cids={}&cids={}", + base_url().await, + did, + cid, + cid + ); + let res = client + .get(&url) + .send() + .await + .expect("Failed to send request"); + + assert_eq!(res.status(), StatusCode::OK); + let content_type = res.headers().get("content-type").unwrap().to_str().unwrap(); + assert!( + content_type.contains("application/vnd.ipld.car"), + "Response should be a CAR file" + ); +} + +#[tokio::test] +async fn test_get_repo_since_returns_partial() { + let client = client(); + let (jwt, did) = create_account_and_login(&client).await; + + let initial_commit_res = client + .get(format!( + "{}/xrpc/com.atproto.sync.getLatestCommit", + base_url().await + )) + .query(&[("did", did.as_str())]) + .send() + .await + .expect("Failed to get initial commit"); + let initial_body: Value = initial_commit_res.json().await.unwrap(); + let initial_rev = initial_body["rev"].as_str().unwrap(); + + let full_repo_res = client + .get(format!( + "{}/xrpc/com.atproto.sync.getRepo", + base_url().await + )) + .query(&[("did", did.as_str())]) + .send() + .await + .expect("Failed to get full repo"); + assert_eq!(full_repo_res.status(), StatusCode::OK); + let full_repo_bytes = full_repo_res.bytes().await.unwrap(); + let full_repo_size = full_repo_bytes.len(); + + create_post(&client, &did, &jwt, "Test post for since param").await; + + let partial_repo_res = client + .get(format!( + "{}/xrpc/com.atproto.sync.getRepo", + base_url().await + )) + .query(&[("did", did.as_str()), ("since", initial_rev)]) + .send() + .await + .expect("Failed to get partial repo"); + assert_eq!(partial_repo_res.status(), StatusCode::OK); + let partial_repo_bytes = partial_repo_res.bytes().await.unwrap(); + let partial_repo_size = partial_repo_bytes.len(); + + assert!( + partial_repo_size < full_repo_size, + "Partial export (since={}) should be smaller than full export: {} vs {}", + initial_rev, + partial_repo_size, + full_repo_size + ); +} + +#[tokio::test] +async fn test_list_blobs_takendown_returns_error() { + let client = client(); + let (_, did) = create_account_and_login(&client).await; + + set_account_takedown(&did, Some("test-takedown-ref")).await; + + let res = client + .get(format!( + "{}/xrpc/com.atproto.sync.listBlobs", + base_url().await + )) + .query(&[("did", did.as_str())]) + .send() + .await + .expect("Failed to send request"); + + assert_eq!(res.status(), StatusCode::BAD_REQUEST); + let body: Value = res.json().await.expect("Response was not valid JSON"); + assert_eq!(body["error"], "RepoTakendown"); +} + +#[tokio::test] +async fn test_get_record_takendown_returns_error() { + let client = client(); + let (jwt, did) = create_account_and_login(&client).await; + + let (uri, _cid) = create_post(&client, &did, &jwt, "Test post").await; + let parts: Vec<&str> = uri.split('/').collect(); + let collection = parts[parts.len() - 2]; + let rkey = parts[parts.len() - 1]; + + set_account_takedown(&did, Some("test-takedown-ref")).await; + + let res = client + .get(format!( + "{}/xrpc/com.atproto.sync.getRecord", + base_url().await + )) + .query(&[ + ("did", did.as_str()), + ("collection", collection), + ("rkey", rkey), + ]) + .send() + .await + .expect("Failed to send request"); + + assert_eq!(res.status(), StatusCode::BAD_REQUEST); + let body: Value = res.json().await.expect("Response was not valid JSON"); + assert_eq!(body["error"], "RepoTakendown"); +} diff --git a/tests/sync_deprecated.rs b/tests/sync_deprecated.rs index e1390c5..921374a 100644 --- a/tests/sync_deprecated.rs +++ b/tests/sync_deprecated.rs @@ -138,7 +138,7 @@ async fn test_get_checkout_comprehensive() { "CAR file should have at least header length" ); for i in 0..4 { - tokio::time::sleep(std::time::Duration::from_millis(50)).await; + tokio::time::sleep(std::time::Duration::from_millis(100)).await; create_post(&client, &did, &jwt, &format!("Checkout post {}", i)).await; } let multi_res = client diff --git a/tests/sync_repo.rs b/tests/sync_repo.rs index 1807864..43483f1 100644 --- a/tests/sync_repo.rs +++ b/tests/sync_repo.rs @@ -39,7 +39,7 @@ async fn test_get_latest_commit_not_found() { .send() .await .expect("Failed to send request"); - assert_eq!(res.status(), StatusCode::NOT_FOUND); + assert_eq!(res.status(), StatusCode::BAD_REQUEST); let body: Value = res.json().await.expect("Response was not valid JSON"); assert_eq!(body["error"], "RepoNotFound"); } @@ -106,25 +106,19 @@ async fn test_list_repos_with_limit() { #[tokio::test] async fn test_list_repos_pagination() { let client = client(); - let _ = create_account_and_login(&client).await; - let _ = create_account_and_login(&client).await; - let _ = create_account_and_login(&client).await; - let params = [("limit", "1")]; - let res = client - .get(format!( - "{}/xrpc/com.atproto.sync.listRepos", - base_url().await - )) - .query(¶ms) - .send() - .await - .expect("Failed to send request"); - assert_eq!(res.status(), StatusCode::OK); - let body: Value = res.json().await.expect("Response was not valid JSON"); - let repos = body["repos"].as_array().unwrap(); - assert_eq!(repos.len(), 1); - if let Some(cursor) = body["cursor"].as_str() { - let params = [("limit", "1"), ("cursor", cursor)]; + let (_, did1) = create_account_and_login(&client).await; + let (_, did2) = create_account_and_login(&client).await; + let (_, did3) = create_account_and_login(&client).await; + let our_dids: std::collections::HashSet = [did1, did2, did3].into_iter().collect(); + let mut all_dids_seen: std::collections::HashSet = std::collections::HashSet::new(); + let mut cursor: Option = None; + let mut page_count = 0; + let max_pages = 100; + loop { + let mut params: Vec<(&str, String)> = vec![("limit".into(), "10".into())]; + if let Some(ref c) = cursor { + params.push(("cursor", c.clone())); + } let res = client .get(format!( "{}/xrpc/com.atproto.sync.listRepos", @@ -136,9 +130,28 @@ async fn test_list_repos_pagination() { .expect("Failed to send request"); assert_eq!(res.status(), StatusCode::OK); let body: Value = res.json().await.expect("Response was not valid JSON"); - let repos2 = body["repos"].as_array().unwrap(); - assert_eq!(repos2.len(), 1); - assert_ne!(repos[0]["did"], repos2[0]["did"]); + let repos = body["repos"].as_array().unwrap(); + for repo in repos { + let did = repo["did"].as_str().unwrap().to_string(); + assert!( + !all_dids_seen.contains(&did), + "Pagination returned duplicate DID: {}", + did + ); + all_dids_seen.insert(did); + } + cursor = body["cursor"].as_str().map(String::from); + page_count += 1; + if cursor.is_none() || page_count >= max_pages { + break; + } + } + for did in &our_dids { + assert!( + all_dids_seen.contains(did), + "Our created DID {} was not found in paginated results", + did + ); } } @@ -176,7 +189,7 @@ async fn test_get_repo_status_not_found() { .send() .await .expect("Failed to send request"); - assert_eq!(res.status(), StatusCode::NOT_FOUND); + assert_eq!(res.status(), StatusCode::BAD_REQUEST); let body: Value = res.json().await.expect("Response was not valid JSON"); assert_eq!(body["error"], "RepoNotFound"); } @@ -270,7 +283,7 @@ async fn test_get_repo_not_found() { .send() .await .expect("Failed to send request"); - assert_eq!(res.status(), StatusCode::NOT_FOUND); + assert_eq!(res.status(), StatusCode::BAD_REQUEST); let body: Value = res.json().await.expect("Response was not valid JSON"); assert_eq!(body["error"], "RepoNotFound"); } @@ -397,7 +410,7 @@ async fn test_get_blocks_not_found() { .send() .await .expect("Failed to send request"); - assert_eq!(res.status(), StatusCode::NOT_FOUND); + assert_eq!(res.status(), StatusCode::BAD_REQUEST); } #[tokio::test] @@ -536,7 +549,7 @@ async fn test_sync_repo_export_lifecycle() { .expect("Failed to create profile"); assert_eq!(profile_res.status(), StatusCode::OK); for i in 0..3 { - tokio::time::sleep(std::time::Duration::from_millis(50)).await; + tokio::time::sleep(std::time::Duration::from_millis(100)).await; create_post(&client, &did, &jwt, &format!("Export test post {}", i)).await; } let blob_data = b"blob data for sync export test";