diff --git a/.sqlx/query-0349b2af4d21105029ecc2bf81ad7be9987ec17ac088730596849d7a45558e57.json b/.sqlx/query-0349b2af4d21105029ecc2bf81ad7be9987ec17ac088730596849d7a45558e57.json deleted file mode 100644 --- a/.sqlx/query-0349b2af4d21105029ecc2bf81ad7be9987ec17ac088730596849d7a45558e57.json +++ /dev/null @@ -1,113 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT seq, did, created_at, event_type as \"event_type: RepoEventType\", commit_cid, prev_cid, prev_data_cid,\n ops, blobs, block_cids, block_data, blocks_cids, handle, active, status, rev\n FROM repo_seq\n WHERE seq > $1\n ORDER BY seq ASC\n LIMIT $2", - "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: RepoEventType", - "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": "block_cids", - "type_info": "ByteaArray" - }, - { - "ordinal": 10, - "name": "block_data", - "type_info": "ByteaArray" - }, - { - "ordinal": 11, - "name": "blocks_cids", - "type_info": "TextArray" - }, - { - "ordinal": 12, - "name": "handle", - "type_info": "Text" - }, - { - "ordinal": 13, - "name": "active", - "type_info": "Bool" - }, - { - "ordinal": 14, - "name": "status", - "type_info": "Text" - }, - { - "ordinal": 15, - "name": "rev", - "type_info": "Text" - } - ], - "parameters": { - "Left": [ - "Int8", - "Int8" - ] - }, - "nullable": [ - false, - false, - false, - false, - true, - true, - true, - true, - true, - true, - true, - true, - true, - true, - true, - true - ] - }, - "hash": "0349b2af4d21105029ecc2bf81ad7be9987ec17ac088730596849d7a45558e57" -} diff --git a/.sqlx/query-09142b5a6b91fe4fd795fb6c85af5461e2621f6c3545b6d26e4618d813904b75.json b/.sqlx/query-09142b5a6b91fe4fd795fb6c85af5461e2621f6c3545b6d26e4618d813904b75.json new file mode 100644 --- /dev/null +++ b/.sqlx/query-09142b5a6b91fe4fd795fb6c85af5461e2621f6c3545b6d26e4618d813904b75.json @@ -0,0 +1,20 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT nextval('firehose_seq') as \"seq!\"", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "seq!", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + null + ] + }, + "hash": "09142b5a6b91fe4fd795fb6c85af5461e2621f6c3545b6d26e4618d813904b75" +} diff --git a/.sqlx/query-0b895d105363a60040323c5c47bc897905af3da980d0dc1b3b60bbdcdd90ea7a.json b/.sqlx/query-0b895d105363a60040323c5c47bc897905af3da980d0dc1b3b60bbdcdd90ea7a.json new file mode 100644 --- /dev/null +++ b/.sqlx/query-0b895d105363a60040323c5c47bc897905af3da980d0dc1b3b60bbdcdd90ea7a.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE repo_seq r\n SET seq = $1::bigint + (o.ord - 1)\n FROM unnest($2::bigint[]) WITH ORDINALITY AS o(id, ord)\n WHERE r.id = o.id", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Int8", + "Int8Array" + ] + }, + "nullable": [] + }, + "hash": "0b895d105363a60040323c5c47bc897905af3da980d0dc1b3b60bbdcdd90ea7a" +} diff --git a/.sqlx/query-122d79e006c0e11214e3cf11cfb98bdff6fbf156893b482572532ebcd8979e61.json b/.sqlx/query-122d79e006c0e11214e3cf11cfb98bdff6fbf156893b482572532ebcd8979e61.json new file mode 100644 --- /dev/null +++ b/.sqlx/query-122d79e006c0e11214e3cf11cfb98bdff6fbf156893b482572532ebcd8979e61.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "\n DELETE FROM repo_seq\n WHERE did = $1\n AND id <> (SELECT id FROM repo_seq WHERE did = $1 ORDER BY id DESC LIMIT 1)\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [] + }, + "hash": "122d79e006c0e11214e3cf11cfb98bdff6fbf156893b482572532ebcd8979e61" +} diff --git a/.sqlx/query-2c3c0740793f29b9bd9e0f81c2372d427dee091e20da8f71ae1e5f2769aa985d.json b/.sqlx/query-2c3c0740793f29b9bd9e0f81c2372d427dee091e20da8f71ae1e5f2769aa985d.json new file mode 100644 --- /dev/null +++ b/.sqlx/query-2c3c0740793f29b9bd9e0f81c2372d427dee091e20da8f71ae1e5f2769aa985d.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO repo_seq (did, event_type, active, status)\n VALUES ($1, 'account', false, 'deleted')\n RETURNING id\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + false + ] + }, + "hash": "2c3c0740793f29b9bd9e0f81c2372d427dee091e20da8f71ae1e5f2769aa985d" +} diff --git a/.sqlx/query-3cc7247d2cd7e85a6f21d783dd956f8fecc6f63c994014bb4e91504d498b5fd5.json b/.sqlx/query-3cc7247d2cd7e85a6f21d783dd956f8fecc6f63c994014bb4e91504d498b5fd5.json new file mode 100644 --- /dev/null +++ b/.sqlx/query-3cc7247d2cd7e85a6f21d783dd956f8fecc6f63c994014bb4e91504d498b5fd5.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT pg_try_advisory_xact_lock($1) as \"locked!\"", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "locked!", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [ + null + ] + }, + "hash": "3cc7247d2cd7e85a6f21d783dd956f8fecc6f63c994014bb4e91504d498b5fd5" +} diff --git a/.sqlx/query-4702a94384ce938d3f5384bc2106a48500f8749134e6bfd1c619a9151f5d3772.json b/.sqlx/query-4702a94384ce938d3f5384bc2106a48500f8749134e6bfd1c619a9151f5d3772.json deleted file mode 100644 --- a/.sqlx/query-4702a94384ce938d3f5384bc2106a48500f8749134e6bfd1c619a9151f5d3772.json +++ /dev/null @@ -1,112 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT seq, did, created_at, event_type as \"event_type: RepoEventType\", commit_cid, prev_cid, prev_data_cid,\n ops, blobs, block_cids, block_data, blocks_cids, handle, active, status, rev\n FROM repo_seq\n WHERE seq = $1", - "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: RepoEventType", - "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": "block_cids", - "type_info": "ByteaArray" - }, - { - "ordinal": 10, - "name": "block_data", - "type_info": "ByteaArray" - }, - { - "ordinal": 11, - "name": "blocks_cids", - "type_info": "TextArray" - }, - { - "ordinal": 12, - "name": "handle", - "type_info": "Text" - }, - { - "ordinal": 13, - "name": "active", - "type_info": "Bool" - }, - { - "ordinal": 14, - "name": "status", - "type_info": "Text" - }, - { - "ordinal": 15, - "name": "rev", - "type_info": "Text" - } - ], - "parameters": { - "Left": [ - "Int8" - ] - }, - "nullable": [ - false, - false, - false, - false, - true, - true, - true, - true, - true, - true, - true, - true, - true, - true, - true, - true - ] - }, - "hash": "4702a94384ce938d3f5384bc2106a48500f8749134e6bfd1c619a9151f5d3772" -} diff --git a/.sqlx/query-4f3ef746cb5fc0916d3df5f92494c538497d3b9c1cfcc6e44c9b80f2193094f6.json b/.sqlx/query-4f3ef746cb5fc0916d3df5f92494c538497d3b9c1cfcc6e44c9b80f2193094f6.json new file mode 100644 --- /dev/null +++ b/.sqlx/query-4f3ef746cb5fc0916d3df5f92494c538497d3b9c1cfcc6e44c9b80f2193094f6.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO repo_seq (did, event_type, handle)\n VALUES ($1, 'identity', $2)\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [] + }, + "hash": "4f3ef746cb5fc0916d3df5f92494c538497d3b9c1cfcc6e44c9b80f2193094f6" +} diff --git a/.sqlx/query-50b2178cd174af1dc132a22048e52a6af492f4d8025ec512dd9c0c9ba62eb202.json b/.sqlx/query-50b2178cd174af1dc132a22048e52a6af492f4d8025ec512dd9c0c9ba62eb202.json deleted file mode 100644 --- a/.sqlx/query-50b2178cd174af1dc132a22048e52a6af492f4d8025ec512dd9c0c9ba62eb202.json +++ /dev/null @@ -1,23 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n INSERT INTO repo_seq (did, event_type, handle)\n VALUES ($1, 'identity', $2)\n RETURNING seq\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "seq", - "type_info": "Int8" - } - ], - "parameters": { - "Left": [ - "Text", - "Text" - ] - }, - "nullable": [ - false - ] - }, - "hash": "50b2178cd174af1dc132a22048e52a6af492f4d8025ec512dd9c0c9ba62eb202" -} diff --git a/.sqlx/query-5996774d097a484f81353d49e4a99b07c30a003377c924add576bfb48ddde29b.json b/.sqlx/query-5996774d097a484f81353d49e4a99b07c30a003377c924add576bfb48ddde29b.json deleted file mode 100644 --- a/.sqlx/query-5996774d097a484f81353d49e4a99b07c30a003377c924add576bfb48ddde29b.json +++ /dev/null @@ -1,26 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n INSERT INTO repo_seq (did, event_type, commit_cid, rev, block_cids, block_data)\n VALUES ($1, 'sync', $2, $3, $4, $5)\n RETURNING seq\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "seq", - "type_info": "Int8" - } - ], - "parameters": { - "Left": [ - "Text", - "Text", - "Text", - "ByteaArray", - "ByteaArray" - ] - }, - "nullable": [ - false - ] - }, - "hash": "5996774d097a484f81353d49e4a99b07c30a003377c924add576bfb48ddde29b" -} diff --git a/.sqlx/query-623042c0b901e17ff37c5bcc1a4eefe56fc8b996ca8326e75796897d13c9e9ed.json b/.sqlx/query-623042c0b901e17ff37c5bcc1a4eefe56fc8b996ca8326e75796897d13c9e9ed.json deleted file mode 100644 --- a/.sqlx/query-623042c0b901e17ff37c5bcc1a4eefe56fc8b996ca8326e75796897d13c9e9ed.json +++ /dev/null @@ -1,24 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n INSERT INTO repo_seq (did, event_type, active, status)\n VALUES ($1, 'account', $2, $3)\n RETURNING seq\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "seq", - "type_info": "Int8" - } - ], - "parameters": { - "Left": [ - "Text", - "Bool", - "Text" - ] - }, - "nullable": [ - false - ] - }, - "hash": "623042c0b901e17ff37c5bcc1a4eefe56fc8b996ca8326e75796897d13c9e9ed" -} diff --git a/.sqlx/query-714a1c9a73e2fed8182d4ab2fe4a2c33bc10a52e27ed95d3a6641477ae42395e.json b/.sqlx/query-714a1c9a73e2fed8182d4ab2fe4a2c33bc10a52e27ed95d3a6641477ae42395e.json new file mode 100644 --- /dev/null +++ b/.sqlx/query-714a1c9a73e2fed8182d4ab2fe4a2c33bc10a52e27ed95d3a6641477ae42395e.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT id as \"id!\" FROM repo_seq WHERE seq IS NULL ORDER BY id LIMIT $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id!", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [ + false + ] + }, + "hash": "714a1c9a73e2fed8182d4ab2fe4a2c33bc10a52e27ed95d3a6641477ae42395e" +} diff --git a/.sqlx/query-7cdf3b8136782a5b2e4aff91907a680566b4610e6cbf9d9396d31f9bbd99c3b8.json b/.sqlx/query-7cdf3b8136782a5b2e4aff91907a680566b4610e6cbf9d9396d31f9bbd99c3b8.json new file mode 100644 --- /dev/null +++ b/.sqlx/query-7cdf3b8136782a5b2e4aff91907a680566b4610e6cbf9d9396d31f9bbd99c3b8.json @@ -0,0 +1,112 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT seq as \"seq!\", did, created_at, event_type as \"event_type: RepoEventType\", commit_cid, prev_cid, prev_data_cid,\n ops, blobs, block_cids, block_data, blocks_cids, handle, active, status, rev\n FROM repo_seq\n WHERE seq = $1", + "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: RepoEventType", + "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": "block_cids", + "type_info": "ByteaArray" + }, + { + "ordinal": 10, + "name": "block_data", + "type_info": "ByteaArray" + }, + { + "ordinal": 11, + "name": "blocks_cids", + "type_info": "TextArray" + }, + { + "ordinal": 12, + "name": "handle", + "type_info": "Text" + }, + { + "ordinal": 13, + "name": "active", + "type_info": "Bool" + }, + { + "ordinal": 14, + "name": "status", + "type_info": "Text" + }, + { + "ordinal": 15, + "name": "rev", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [ + true, + false, + false, + false, + true, + true, + true, + true, + true, + true, + true, + true, + true, + true, + true, + true + ] + }, + "hash": "7cdf3b8136782a5b2e4aff91907a680566b4610e6cbf9d9396d31f9bbd99c3b8" +} diff --git a/.sqlx/query-800b93a38eef70bafdf0a1b221994f3c19ab562f84d18265376296a592b4c767.json b/.sqlx/query-800b93a38eef70bafdf0a1b221994f3c19ab562f84d18265376296a592b4c767.json new file mode 100644 --- /dev/null +++ b/.sqlx/query-800b93a38eef70bafdf0a1b221994f3c19ab562f84d18265376296a592b4c767.json @@ -0,0 +1,21 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO repo_seq (did, event_type, commit_cid, prev_cid, ops, blobs, block_cids, block_data, rev)\n VALUES ($1, 'commit', $2, $3::TEXT, $4, $5, $6, $7, $8)\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Text", + "Text", + "Jsonb", + "TextArray", + "ByteaArray", + "ByteaArray", + "Text" + ] + }, + "nullable": [] + }, + "hash": "800b93a38eef70bafdf0a1b221994f3c19ab562f84d18265376296a592b4c767" +} diff --git a/.sqlx/query-82276a34f473ff8b9790b78915ae3f72873d4b748cdb47bd2b81edc2f7a6e5a8.json b/.sqlx/query-82276a34f473ff8b9790b78915ae3f72873d4b748cdb47bd2b81edc2f7a6e5a8.json new file mode 100644 --- /dev/null +++ b/.sqlx/query-82276a34f473ff8b9790b78915ae3f72873d4b748cdb47bd2b81edc2f7a6e5a8.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO repo_seq (did, event_type, commit_cid, prev_cid, ops, blobs, block_cids, block_data, prev_data_cid, rev)\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Text", + "Text", + "Text", + "Jsonb", + "TextArray", + "ByteaArray", + "ByteaArray", + "Text", + "Text" + ] + }, + "nullable": [] + }, + "hash": "82276a34f473ff8b9790b78915ae3f72873d4b748cdb47bd2b81edc2f7a6e5a8" +} diff --git a/.sqlx/query-856e458c1c78f030545dc147f76d1b91fa4cc6e178eb51133c52358855ba3f02.json b/.sqlx/query-856e458c1c78f030545dc147f76d1b91fa4cc6e178eb51133c52358855ba3f02.json new file mode 100644 --- /dev/null +++ b/.sqlx/query-856e458c1c78f030545dc147f76d1b91fa4cc6e178eb51133c52358855ba3f02.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM repo_seq WHERE did = $1 AND id <> $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Int8" + ] + }, + "nullable": [] + }, + "hash": "856e458c1c78f030545dc147f76d1b91fa4cc6e178eb51133c52358855ba3f02" +} diff --git a/.sqlx/query-89c8ce80d8c52b4668c34f63e48b51cacc7e8ff300958a192378f1d72de04f9a.json b/.sqlx/query-89c8ce80d8c52b4668c34f63e48b51cacc7e8ff300958a192378f1d72de04f9a.json deleted file mode 100644 --- a/.sqlx/query-89c8ce80d8c52b4668c34f63e48b51cacc7e8ff300958a192378f1d72de04f9a.json +++ /dev/null @@ -1,113 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT seq, did, created_at, event_type as \"event_type: RepoEventType\", commit_cid, prev_cid, prev_data_cid,\n ops, blobs, block_cids, block_data, blocks_cids, handle, active, status, rev\n FROM repo_seq\n WHERE seq > $1\n ORDER BY seq ASC\n LIMIT $2", - "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: RepoEventType", - "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": "block_cids", - "type_info": "ByteaArray" - }, - { - "ordinal": 10, - "name": "block_data", - "type_info": "ByteaArray" - }, - { - "ordinal": 11, - "name": "blocks_cids", - "type_info": "TextArray" - }, - { - "ordinal": 12, - "name": "handle", - "type_info": "Text" - }, - { - "ordinal": 13, - "name": "active", - "type_info": "Bool" - }, - { - "ordinal": 14, - "name": "status", - "type_info": "Text" - }, - { - "ordinal": 15, - "name": "rev", - "type_info": "Text" - } - ], - "parameters": { - "Left": [ - "Int8", - "Int8" - ] - }, - "nullable": [ - false, - false, - false, - false, - true, - true, - true, - true, - true, - true, - true, - true, - true, - true, - true, - true - ] - }, - "hash": "89c8ce80d8c52b4668c34f63e48b51cacc7e8ff300958a192378f1d72de04f9a" -} diff --git a/.sqlx/query-8ccd42b5b1b75f45e094889ee26af8517de28ca687efa81fd9c4fce598338fe9.json b/.sqlx/query-8ccd42b5b1b75f45e094889ee26af8517de28ca687efa81fd9c4fce598338fe9.json new file mode 100644 --- /dev/null +++ b/.sqlx/query-8ccd42b5b1b75f45e094889ee26af8517de28ca687efa81fd9c4fce598338fe9.json @@ -0,0 +1,16 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO repo_seq (did, event_type, active, status)\n VALUES ($1, 'account', $2, $3)\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Bool", + "Text" + ] + }, + "nullable": [] + }, + "hash": "8ccd42b5b1b75f45e094889ee26af8517de28ca687efa81fd9c4fce598338fe9" +} diff --git a/.sqlx/query-957bc81c63510ab502d2f8cea75dafcbdf2c8ebb448c86990d5e404ee5280438.json b/.sqlx/query-957bc81c63510ab502d2f8cea75dafcbdf2c8ebb448c86990d5e404ee5280438.json new file mode 100644 --- /dev/null +++ b/.sqlx/query-957bc81c63510ab502d2f8cea75dafcbdf2c8ebb448c86990d5e404ee5280438.json @@ -0,0 +1,112 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT seq as \"seq!\", did, created_at, event_type as \"event_type: RepoEventType\", commit_cid, prev_cid, prev_data_cid,\n ops, blobs, block_cids, block_data, blocks_cids, handle, active, status, rev\n FROM repo_seq\n WHERE seq > $1\n ORDER BY seq ASC", + "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: RepoEventType", + "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": "block_cids", + "type_info": "ByteaArray" + }, + { + "ordinal": 10, + "name": "block_data", + "type_info": "ByteaArray" + }, + { + "ordinal": 11, + "name": "blocks_cids", + "type_info": "TextArray" + }, + { + "ordinal": 12, + "name": "handle", + "type_info": "Text" + }, + { + "ordinal": 13, + "name": "active", + "type_info": "Bool" + }, + { + "ordinal": 14, + "name": "status", + "type_info": "Text" + }, + { + "ordinal": 15, + "name": "rev", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [ + true, + false, + false, + false, + true, + true, + true, + true, + true, + true, + true, + true, + true, + true, + true, + true + ] + }, + "hash": "957bc81c63510ab502d2f8cea75dafcbdf2c8ebb448c86990d5e404ee5280438" +} diff --git a/.sqlx/query-97dd3cfe72b74b4dac686c030a07f6eb56dabb2df72815f3248760b4e3f7950e.json b/.sqlx/query-97dd3cfe72b74b4dac686c030a07f6eb56dabb2df72815f3248760b4e3f7950e.json deleted file mode 100644 --- a/.sqlx/query-97dd3cfe72b74b4dac686c030a07f6eb56dabb2df72815f3248760b4e3f7950e.json +++ /dev/null @@ -1,31 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n INSERT INTO repo_seq (did, event_type, commit_cid, prev_cid, ops, blobs, block_cids, block_data, prev_data_cid, rev)\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)\n RETURNING seq\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "seq", - "type_info": "Int8" - } - ], - "parameters": { - "Left": [ - "Text", - "Text", - "Text", - "Text", - "Jsonb", - "TextArray", - "ByteaArray", - "ByteaArray", - "Text", - "Text" - ] - }, - "nullable": [ - false - ] - }, - "hash": "97dd3cfe72b74b4dac686c030a07f6eb56dabb2df72815f3248760b4e3f7950e" -} diff --git a/.sqlx/query-9ba7869f29286cfd6b811ffa1e26bda1d637b98b44be66cdc265b1af4cec2d4f.json b/.sqlx/query-9ba7869f29286cfd6b811ffa1e26bda1d637b98b44be66cdc265b1af4cec2d4f.json new file mode 100644 --- /dev/null +++ b/.sqlx/query-9ba7869f29286cfd6b811ffa1e26bda1d637b98b44be66cdc265b1af4cec2d4f.json @@ -0,0 +1,18 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO repo_seq (did, event_type, commit_cid, rev, block_cids, block_data)\n VALUES ($1, 'sync', $2, $3, $4, $5)\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Text", + "Text", + "ByteaArray", + "ByteaArray" + ] + }, + "nullable": [] + }, + "hash": "9ba7869f29286cfd6b811ffa1e26bda1d637b98b44be66cdc265b1af4cec2d4f" +} diff --git a/.sqlx/query-a06e1d9f6f95e4c4c2b98310ebddcc9d963cc033582bf2e945e8bf3a301b4247.json b/.sqlx/query-a06e1d9f6f95e4c4c2b98310ebddcc9d963cc033582bf2e945e8bf3a301b4247.json new file mode 100644 --- /dev/null +++ b/.sqlx/query-a06e1d9f6f95e4c4c2b98310ebddcc9d963cc033582bf2e945e8bf3a301b4247.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT pg_advisory_xact_lock($1)", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "pg_advisory_xact_lock", + "type_info": "Void" + } + ], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [ + null + ] + }, + "hash": "a06e1d9f6f95e4c4c2b98310ebddcc9d963cc033582bf2e945e8bf3a301b4247" +} diff --git a/.sqlx/query-a0ea3630adb9e411d94d88084df6b05d7fc3ba8c306146fc932b736fe342eceb.json b/.sqlx/query-a0ea3630adb9e411d94d88084df6b05d7fc3ba8c306146fc932b736fe342eceb.json deleted file mode 100644 --- a/.sqlx/query-a0ea3630adb9e411d94d88084df6b05d7fc3ba8c306146fc932b736fe342eceb.json +++ /dev/null @@ -1,22 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n INSERT INTO repo_seq (did, event_type, active, status)\n VALUES ($1, 'account', false, 'deleted')\n RETURNING seq\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "seq", - "type_info": "Int8" - } - ], - "parameters": { - "Left": [ - "Text" - ] - }, - "nullable": [ - false - ] - }, - "hash": "a0ea3630adb9e411d94d88084df6b05d7fc3ba8c306146fc932b736fe342eceb" -} diff --git a/.sqlx/query-a325862f897484bb1a8863fd7a1bc3fd17ab20b6e30498f9900daf80609c413d.json b/.sqlx/query-a325862f897484bb1a8863fd7a1bc3fd17ab20b6e30498f9900daf80609c413d.json deleted file mode 100644 --- a/.sqlx/query-a325862f897484bb1a8863fd7a1bc3fd17ab20b6e30498f9900daf80609c413d.json +++ /dev/null @@ -1,29 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n INSERT INTO repo_seq (did, event_type, commit_cid, prev_cid, ops, blobs, block_cids, block_data, rev)\n VALUES ($1, 'commit', $2, $3::TEXT, $4, $5, $6, $7, $8)\n RETURNING seq\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "seq", - "type_info": "Int8" - } - ], - "parameters": { - "Left": [ - "Text", - "Text", - "Text", - "Jsonb", - "TextArray", - "ByteaArray", - "ByteaArray", - "Text" - ] - }, - "nullable": [ - false - ] - }, - "hash": "a325862f897484bb1a8863fd7a1bc3fd17ab20b6e30498f9900daf80609c413d" -} diff --git a/.sqlx/query-ad784aff9bb5c1f19f90a3b1c3b3694a3e1aa889b0fe3e55cfebb7af8c620256.json b/.sqlx/query-ad784aff9bb5c1f19f90a3b1c3b3694a3e1aa889b0fe3e55cfebb7af8c620256.json deleted file mode 100644 --- a/.sqlx/query-ad784aff9bb5c1f19f90a3b1c3b3694a3e1aa889b0fe3e55cfebb7af8c620256.json +++ /dev/null @@ -1,113 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT seq, did, created_at, event_type as \"event_type: RepoEventType\", commit_cid, prev_cid, prev_data_cid,\n ops, blobs, block_cids, block_data, blocks_cids, handle, active, status, rev\n FROM repo_seq\n WHERE seq > $1 AND seq < $2\n ORDER BY seq ASC", - "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: RepoEventType", - "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": "block_cids", - "type_info": "ByteaArray" - }, - { - "ordinal": 10, - "name": "block_data", - "type_info": "ByteaArray" - }, - { - "ordinal": 11, - "name": "blocks_cids", - "type_info": "TextArray" - }, - { - "ordinal": 12, - "name": "handle", - "type_info": "Text" - }, - { - "ordinal": 13, - "name": "active", - "type_info": "Bool" - }, - { - "ordinal": 14, - "name": "status", - "type_info": "Text" - }, - { - "ordinal": 15, - "name": "rev", - "type_info": "Text" - } - ], - "parameters": { - "Left": [ - "Int8", - "Int8" - ] - }, - "nullable": [ - false, - false, - false, - false, - true, - true, - true, - true, - true, - true, - true, - true, - true, - true, - true, - true - ] - }, - "hash": "ad784aff9bb5c1f19f90a3b1c3b3694a3e1aa889b0fe3e55cfebb7af8c620256" -} diff --git a/.sqlx/query-b6d6548acb89d6384cd226f6ed0d66de27fde3af24b4a7a3fce7e098812e38a5.json b/.sqlx/query-b6d6548acb89d6384cd226f6ed0d66de27fde3af24b4a7a3fce7e098812e38a5.json deleted file mode 100644 --- a/.sqlx/query-b6d6548acb89d6384cd226f6ed0d66de27fde3af24b4a7a3fce7e098812e38a5.json +++ /dev/null @@ -1,15 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "DELETE FROM repo_seq WHERE did = $1 AND seq != $2", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Text", - "Int8" - ] - }, - "nullable": [] - }, - "hash": "b6d6548acb89d6384cd226f6ed0d66de27fde3af24b4a7a3fce7e098812e38a5" -} diff --git a/.sqlx/query-bd0407818c8bdae1f1024777951b08cb9c386fe751722e1caf6ad243e271697a.json b/.sqlx/query-bd0407818c8bdae1f1024777951b08cb9c386fe751722e1caf6ad243e271697a.json new file mode 100644 --- /dev/null +++ b/.sqlx/query-bd0407818c8bdae1f1024777951b08cb9c386fe751722e1caf6ad243e271697a.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT setval('firehose_seq', $1)", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "setval", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [ + null + ] + }, + "hash": "bd0407818c8bdae1f1024777951b08cb9c386fe751722e1caf6ad243e271697a" +} diff --git a/.sqlx/query-c47746c256fd8c9d386551f3bd270e9a96fecdb713642d07d60fb4aeb02c8426.json b/.sqlx/query-c47746c256fd8c9d386551f3bd270e9a96fecdb713642d07d60fb4aeb02c8426.json deleted file mode 100644 --- a/.sqlx/query-c47746c256fd8c9d386551f3bd270e9a96fecdb713642d07d60fb4aeb02c8426.json +++ /dev/null @@ -1,112 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT seq, did, created_at, event_type as \"event_type: RepoEventType\", commit_cid, prev_cid, prev_data_cid,\n ops, blobs, block_cids, block_data, blocks_cids, handle, active, status, rev\n FROM repo_seq\n WHERE seq > $1\n ORDER BY seq ASC", - "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: RepoEventType", - "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": "block_cids", - "type_info": "ByteaArray" - }, - { - "ordinal": 10, - "name": "block_data", - "type_info": "ByteaArray" - }, - { - "ordinal": 11, - "name": "blocks_cids", - "type_info": "TextArray" - }, - { - "ordinal": 12, - "name": "handle", - "type_info": "Text" - }, - { - "ordinal": 13, - "name": "active", - "type_info": "Bool" - }, - { - "ordinal": 14, - "name": "status", - "type_info": "Text" - }, - { - "ordinal": 15, - "name": "rev", - "type_info": "Text" - } - ], - "parameters": { - "Left": [ - "Int8" - ] - }, - "nullable": [ - false, - false, - false, - false, - true, - true, - true, - true, - true, - true, - true, - true, - true, - true, - true, - true - ] - }, - "hash": "c47746c256fd8c9d386551f3bd270e9a96fecdb713642d07d60fb4aeb02c8426" -} diff --git a/.sqlx/query-d4eacb9bca838dd18dcfd8ca39b83ef50968bfaaabc213536c049cc222ffe9c8.json b/.sqlx/query-d4eacb9bca838dd18dcfd8ca39b83ef50968bfaaabc213536c049cc222ffe9c8.json new file mode 100644 --- /dev/null +++ b/.sqlx/query-d4eacb9bca838dd18dcfd8ca39b83ef50968bfaaabc213536c049cc222ffe9c8.json @@ -0,0 +1,12 @@ +{ + "db_name": "PostgreSQL", + "query": "NOTIFY repo_updates", + "describe": { + "columns": [], + "parameters": { + "Left": [] + }, + "nullable": [] + }, + "hash": "d4eacb9bca838dd18dcfd8ca39b83ef50968bfaaabc213536c049cc222ffe9c8" +} diff --git a/.sqlx/query-da2770e44c0ec21756ce4b6b0c57abff4f38cd8c680c5ced7d3b35f3c1f58da7.json b/.sqlx/query-da2770e44c0ec21756ce4b6b0c57abff4f38cd8c680c5ced7d3b35f3c1f58da7.json new file mode 100644 --- /dev/null +++ b/.sqlx/query-da2770e44c0ec21756ce4b6b0c57abff4f38cd8c680c5ced7d3b35f3c1f58da7.json @@ -0,0 +1,113 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT seq as \"seq!\", did, created_at, event_type as \"event_type: RepoEventType\", commit_cid, prev_cid, prev_data_cid,\n ops, blobs, block_cids, block_data, blocks_cids, handle, active, status, rev\n FROM repo_seq\n WHERE seq > $1\n ORDER BY seq ASC\n LIMIT $2", + "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: RepoEventType", + "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": "block_cids", + "type_info": "ByteaArray" + }, + { + "ordinal": 10, + "name": "block_data", + "type_info": "ByteaArray" + }, + { + "ordinal": 11, + "name": "blocks_cids", + "type_info": "TextArray" + }, + { + "ordinal": 12, + "name": "handle", + "type_info": "Text" + }, + { + "ordinal": 13, + "name": "active", + "type_info": "Bool" + }, + { + "ordinal": 14, + "name": "status", + "type_info": "Text" + }, + { + "ordinal": 15, + "name": "rev", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Int8", + "Int8" + ] + }, + "nullable": [ + true, + false, + false, + false, + true, + true, + true, + true, + true, + true, + true, + true, + true, + true, + true, + true + ] + }, + "hash": "da2770e44c0ec21756ce4b6b0c57abff4f38cd8c680c5ced7d3b35f3c1f58da7" +} diff --git a/.sqlx/query-dfe011b2d2bafbbbbea508a4977fc34bb1be4cf2612296df23763e6e5841c33e.json b/.sqlx/query-dfe011b2d2bafbbbbea508a4977fc34bb1be4cf2612296df23763e6e5841c33e.json new file mode 100644 --- /dev/null +++ b/.sqlx/query-dfe011b2d2bafbbbbea508a4977fc34bb1be4cf2612296df23763e6e5841c33e.json @@ -0,0 +1,113 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT seq as \"seq!\", did, created_at, event_type as \"event_type: RepoEventType\", commit_cid, prev_cid, prev_data_cid,\n ops, blobs, block_cids, block_data, blocks_cids, handle, active, status, rev\n FROM repo_seq\n WHERE seq > $1 AND seq < $2\n ORDER BY seq ASC", + "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: RepoEventType", + "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": "block_cids", + "type_info": "ByteaArray" + }, + { + "ordinal": 10, + "name": "block_data", + "type_info": "ByteaArray" + }, + { + "ordinal": 11, + "name": "blocks_cids", + "type_info": "TextArray" + }, + { + "ordinal": 12, + "name": "handle", + "type_info": "Text" + }, + { + "ordinal": 13, + "name": "active", + "type_info": "Bool" + }, + { + "ordinal": 14, + "name": "status", + "type_info": "Text" + }, + { + "ordinal": 15, + "name": "rev", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Int8", + "Int8" + ] + }, + "nullable": [ + true, + false, + false, + false, + true, + true, + true, + true, + true, + true, + true, + true, + true, + true, + true, + true + ] + }, + "hash": "dfe011b2d2bafbbbbea508a4977fc34bb1be4cf2612296df23763e6e5841c33e" +} diff --git a/.sqlx/query-f16d41b41a5388997e6ed408da55056173340ed12b60b1a956de87f73f9d0824.json b/.sqlx/query-f16d41b41a5388997e6ed408da55056173340ed12b60b1a956de87f73f9d0824.json new file mode 100644 --- /dev/null +++ b/.sqlx/query-f16d41b41a5388997e6ed408da55056173340ed12b60b1a956de87f73f9d0824.json @@ -0,0 +1,113 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT seq as \"seq!\", did, created_at, event_type as \"event_type: RepoEventType\", commit_cid, prev_cid, prev_data_cid,\n ops, blobs, block_cids, block_data, blocks_cids, handle, active, status, rev\n FROM repo_seq\n WHERE seq > $1\n ORDER BY seq ASC\n LIMIT $2", + "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: RepoEventType", + "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": "block_cids", + "type_info": "ByteaArray" + }, + { + "ordinal": 10, + "name": "block_data", + "type_info": "ByteaArray" + }, + { + "ordinal": 11, + "name": "blocks_cids", + "type_info": "TextArray" + }, + { + "ordinal": 12, + "name": "handle", + "type_info": "Text" + }, + { + "ordinal": 13, + "name": "active", + "type_info": "Bool" + }, + { + "ordinal": 14, + "name": "status", + "type_info": "Text" + }, + { + "ordinal": 15, + "name": "rev", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Int8", + "Int8" + ] + }, + "nullable": [ + true, + false, + false, + false, + true, + true, + true, + true, + true, + true, + true, + true, + true, + true, + true, + true + ] + }, + "hash": "f16d41b41a5388997e6ed408da55056173340ed12b60b1a956de87f73f9d0824" +} diff --git a/migrations/20260529_firehose_outbox_sequencing.sql b/migrations/20260529_firehose_outbox_sequencing.sql new file mode 100644 --- /dev/null +++ b/migrations/20260529_firehose_outbox_sequencing.sql @@ -0,0 +1,15 @@ +ALTER TABLE repo_seq ADD COLUMN id BIGSERIAL; + +ALTER TABLE repo_seq DROP CONSTRAINT repo_seq_pkey; +ALTER TABLE repo_seq ADD PRIMARY KEY (id); + +ALTER TABLE repo_seq ALTER COLUMN seq DROP DEFAULT; +ALTER TABLE repo_seq ALTER COLUMN seq DROP NOT NULL; + +DROP INDEX IF EXISTS idx_repo_seq_seq; +ALTER TABLE repo_seq ADD CONSTRAINT repo_seq_seq_key UNIQUE (seq); + +CREATE SEQUENCE IF NOT EXISTS firehose_seq; +SELECT setval('firehose_seq', (SELECT COALESCE(MAX(seq), 0) + 1 FROM repo_seq), false); + +CREATE INDEX idx_repo_seq_unsequenced ON repo_seq (id) WHERE seq IS NULL; diff --git a/crates/tranquil-db-traits/src/repo.rs b/crates/tranquil-db-traits/src/repo.rs --- a/crates/tranquil-db-traits/src/repo.rs +++ b/crates/tranquil-db-traits/src/repo.rs @@ -319,7 +319,6 @@ #[derive(Debug, Clone)] pub struct ApplyCommitResult { - pub seq: i64, pub is_account_active: bool, } @@ -445,19 +444,15 @@ async fn count_user_blocks(&self, user_id: Uuid) -> Result; - async fn insert_commit_event(&self, data: &CommitEventData) -> Result; + async fn insert_commit_event(&self, data: &CommitEventData) -> Result<(), DbError>; async fn insert_identity_event( &self, did: &Did, handle: Option<&Handle>, - ) -> Result; + ) -> Result<(), DbError>; - async fn insert_account_event( - &self, - did: &Did, - status: AccountStatus, - ) -> Result; + async fn insert_account_event(&self, did: &Did, status: AccountStatus) -> Result<(), DbError>; async fn insert_sync_event( &self, @@ -465,7 +460,7 @@ commit_cid: &CidLink, rev: Option<&str>, commit_bytes: &[u8], - ) -> Result; + ) -> Result<(), DbError>; async fn insert_genesis_commit_event( &self, @@ -475,13 +470,17 @@ rev: &str, commit_bytes: &[u8], mst_root_bytes: &[u8], - ) -> Result; - - async fn delete_sequences_except( - &self, - did: &Did, - keep_seq: SequenceNumber, ) -> Result<(), DbError>; + + async fn purge_did_events_keeping_latest(&self, did: &Did) -> Result<(), DbError>; + + async fn assign_pending_sequences(&self) -> Result { + Ok(0) + } + + async fn flush_pending_sequences(&self) -> Result<(), DbError> { + Ok(()) + } async fn prune_events_older_than(&self, cutoff: DateTime) -> Result; @@ -526,8 +525,6 @@ async fn get_repo_root_cid_by_user_id(&self, user_id: Uuid) -> Result, DbError>; - async fn notify_update(&self, seq: SequenceNumber) -> Result<(), DbError>; - async fn import_repo_data( &self, user_id: Uuid, @@ -563,5 +560,5 @@ #[async_trait] pub trait RepoEventReceiver: Send { - async fn recv(&mut self) -> Option; + async fn recv(&mut self) -> Option<()>; } diff --git a/crates/tranquil-db-traits/src/user.rs b/crates/tranquil-db-traits/src/user.rs --- a/crates/tranquil-db-traits/src/user.rs +++ b/crates/tranquil-db-traits/src/user.rs @@ -551,7 +551,7 @@ limit: i64, ) -> Result, DbError>; - async fn delete_account_with_firehose(&self, user_id: Uuid, did: &Did) -> Result; + async fn delete_account_with_firehose(&self, user_id: Uuid, did: &Did) -> Result<(), DbError>; async fn create_password_account( &self, diff --git a/crates/tranquil-pds/src/repo_ops.rs b/crates/tranquil-pds/src/repo_ops.rs --- a/crates/tranquil-pds/src/repo_ops.rs +++ b/crates/tranquil-pds/src/repo_ops.rs @@ -19,7 +19,6 @@ use std::sync::Arc; use tokio::sync::OwnedMutexGuard; use tracing::{error, warn}; -use tranquil_db_traits::SequenceNumber; use uuid::Uuid; #[derive(Debug)] @@ -776,7 +775,7 @@ state: &AppState, did: &Did, handle: Option<&Handle>, -) -> Result { +) -> Result<(), CommitError> { state .repos .repo @@ -788,7 +787,7 @@ state: &AppState, did: &Did, status: tranquil_db_traits::AccountStatus, -) -> Result { +) -> Result<(), CommitError> { state .repos .repo @@ -801,7 +800,7 @@ did: &Did, commit_cid: &str, rev: Option<&str>, -) -> Result { +) -> Result<(), CommitError> { let cid_link: crate::types::CidLink = commit_cid .parse() .map_err(|_| CommitError::InvalidCid(commit_cid.to_string()))?; @@ -829,7 +828,7 @@ commit_cid: &Cid, mst_root_cid: &Cid, rev: &str, -) -> Result { +) -> Result<(), CommitError> { let commit_cid_link = crate::types::CidLink::from(commit_cid); let mst_root_cid_link = crate::types::CidLink::from(mst_root_cid); let commit_bytes = state diff --git a/crates/tranquil-pds/src/scheduled.rs b/crates/tranquil-pds/src/scheduled.rs --- a/crates/tranquil-pds/src/scheduled.rs +++ b/crates/tranquil-pds/src/scheduled.rs @@ -711,7 +711,7 @@ warn!(storage_key = %key, error = %e, "Failed to delete blob from storage (continuing anyway)"); }); - let _account_seq = user_repo + user_repo .delete_account_with_firehose(user_id, did) .await .context("Failed to delete account")?; diff --git a/crates/tranquil-pds/tests/firehose_inline_blocks.rs b/crates/tranquil-pds/tests/firehose_inline_blocks.rs --- a/crates/tranquil-pds/tests/firehose_inline_blocks.rs +++ b/crates/tranquil-pds/tests/firehose_inline_blocks.rs @@ -66,6 +66,11 @@ let repos = get_test_repos().await; let typed_did = tranquil_types::Did::new(did.clone()).unwrap(); + repos + .repo + .flush_pending_sequences() + .await + .expect("flush_pending_sequences"); let events = repos .repo .get_events_since_seq(SequenceNumber::ZERO, None) @@ -142,18 +147,14 @@ let cid_link: CidLink = (&commit_cid).into(); let rev = "3kabcdefghij2"; - let seq = repos + let baseline = repos.repo.get_max_seq().await.expect("get_max_seq"); + repos .repo .insert_sync_event(&did, &cid_link, Some(rev), &commit_bytes) .await .expect("insert_sync_event"); - let event = repos - .repo - .get_event_by_seq(seq) - .await - .expect("get_event_by_seq") - .expect("event present"); + let event = sequenced_event_for_did(repos, baseline, &did).await; assert_eq!(event.event_type, RepoEventType::Sync); let blocks = event @@ -193,7 +194,8 @@ let mst_link: CidLink = (&mst_root_cid).into(); let rev = "3kabcdefghij3"; - let seq = repos + let baseline = repos.repo.get_max_seq().await.expect("get_max_seq"); + repos .repo .insert_genesis_commit_event( &did, @@ -206,12 +208,7 @@ .await .expect("insert_genesis_commit_event"); - let event = repos - .repo - .get_event_by_seq(seq) - .await - .expect("get_event_by_seq") - .expect("event present"); + let event = sequenced_event_for_did(repos, baseline, &did).await; assert_eq!(event.event_type, RepoEventType::Commit); let blocks = event @@ -329,6 +326,11 @@ let repos = get_test_repos().await; let typed_did = tranquil_types::Did::new(did.clone()).unwrap(); + repos + .repo + .flush_pending_sequences() + .await + .expect("flush_pending_sequences"); let events = repos .repo .get_events_since_seq(SequenceNumber::ZERO, None) diff --git a/crates/tranquil-pds/tests/firehose_validation.rs b/crates/tranquil-pds/tests/firehose_validation.rs --- a/crates/tranquil-pds/tests/firehose_validation.rs +++ b/crates/tranquil-pds/tests/firehose_validation.rs @@ -801,7 +801,7 @@ tokio::time::sleep(std::time::Duration::from_millis(100)).await; let repos = get_test_repos().await; - let max_seq = repos.repo.get_max_seq().await.unwrap().as_i64(); + let max_seq = flushed_max_seq(repos).await.as_i64(); let outdated_cursor = (max_seq - 100).max(1); let url = format!( "ws://127.0.0.1:{}/xrpc/com.atproto.sync.subscribeRepos?cursor={}", diff --git a/crates/tranquil-pds/tests/mst_firehose_e2e.rs b/crates/tranquil-pds/tests/mst_firehose_e2e.rs --- a/crates/tranquil-pds/tests/mst_firehose_e2e.rs +++ b/crates/tranquil-pds/tests/mst_firehose_e2e.rs @@ -223,7 +223,7 @@ let (token, did) = create_account_and_login(&client).await; let repos = get_test_repos().await; - let cursor = repos.repo.get_max_seq().await.unwrap().as_i64(); + let cursor = flushed_max_seq(repos).await.as_i64(); let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await; tokio::time::sleep(Duration::from_millis(100)).await; @@ -298,7 +298,7 @@ let (token, did) = create_account_and_login(&client).await; let repos = get_test_repos().await; - let cursor = repos.repo.get_max_seq().await.unwrap().as_i64(); + let cursor = flushed_max_seq(repos).await.as_i64(); let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await; tokio::time::sleep(Duration::from_millis(100)).await; @@ -340,7 +340,7 @@ create_record(&client, &token, &did, &rkey_for(i), "pre").await; } - let resume_cursor = repos.repo.get_max_seq().await.unwrap().as_i64(); + let resume_cursor = flushed_max_seq(repos).await.as_i64(); for i in 5..12 { create_record(&client, &token, &did, &rkey_for(i), "post").await; @@ -371,7 +371,7 @@ let client = client(); let (token, did) = create_account_and_login(&client).await; let repos = get_test_repos().await; - let cursor = repos.repo.get_max_seq().await.unwrap().as_i64(); + let cursor = flushed_max_seq(repos).await.as_i64(); let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await; tokio::time::sleep(Duration::from_millis(100)).await; @@ -417,7 +417,7 @@ let client = client(); let (token, did) = create_account_and_login(&client).await; let repos = get_test_repos().await; - let cursor = repos.repo.get_max_seq().await.unwrap().as_i64(); + let cursor = flushed_max_seq(repos).await.as_i64(); let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await; tokio::time::sleep(Duration::from_millis(100)).await; diff --git a/crates/tranquil-pds/tests/mst_inductive_firehose.rs b/crates/tranquil-pds/tests/mst_inductive_firehose.rs --- a/crates/tranquil-pds/tests/mst_inductive_firehose.rs +++ b/crates/tranquil-pds/tests/mst_inductive_firehose.rs @@ -247,6 +247,11 @@ async fn our_commit_events(did: &str) -> Vec { let repos = get_test_repos().await; let typed_did = Did::new(did.to_string()).unwrap(); + repos + .repo + .flush_pending_sequences() + .await + .expect("flush_pending_sequences"); let events = repos .repo .get_events_since_seq(SequenceNumber::ZERO, None) diff --git a/crates/tranquil-pds/tests/mst_integrity.rs b/crates/tranquil-pds/tests/mst_integrity.rs --- a/crates/tranquil-pds/tests/mst_integrity.rs +++ b/crates/tranquil-pds/tests/mst_integrity.rs @@ -216,6 +216,11 @@ let repos = get_test_repos().await; let typed_did = Did::new(did.clone()).unwrap(); + repos + .repo + .flush_pending_sequences() + .await + .expect("flush_pending_sequences"); let events = repos .repo .get_events_since_seq(SequenceNumber::ZERO, None) @@ -253,7 +258,7 @@ let (token, did) = create_account_and_login(&client).await; let repos = get_test_repos().await; - let cursor = repos.repo.get_max_seq().await.unwrap().as_i64(); + let cursor = flushed_max_seq(repos).await.as_i64(); let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await; tokio::time::sleep(Duration::from_millis(100)).await; @@ -284,7 +289,7 @@ let (token, did) = create_account_and_login(&client).await; let repos = get_test_repos().await; - let cursor = repos.repo.get_max_seq().await.unwrap().as_i64(); + let cursor = flushed_max_seq(repos).await.as_i64(); let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await; tokio::time::sleep(Duration::from_millis(100)).await; diff --git a/crates/tranquil-pds/tests/mst_property_fuzz.rs b/crates/tranquil-pds/tests/mst_property_fuzz.rs --- a/crates/tranquil-pds/tests/mst_property_fuzz.rs +++ b/crates/tranquil-pds/tests/mst_property_fuzz.rs @@ -267,6 +267,11 @@ let repos = get_test_repos().await; let typed_did = Did::new(did.clone()).unwrap(); + repos + .repo + .flush_pending_sequences() + .await + .expect("flush_pending_sequences"); let events = repos .repo .get_events_since_seq(SequenceNumber::ZERO, None) diff --git a/crates/tranquil-pds/tests/repo_lifecycle.rs b/crates/tranquil-pds/tests/repo_lifecycle.rs --- a/crates/tranquil-pds/tests/repo_lifecycle.rs +++ b/crates/tranquil-pds/tests/repo_lifecycle.rs @@ -59,7 +59,7 @@ let (token, did) = create_account_and_login(&client).await; let repos = get_test_repos().await; - let cursor = repos.repo.get_max_seq().await.unwrap().as_i64(); + let cursor = flushed_max_seq(repos).await.as_i64(); let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await; tokio::time::sleep(std::time::Duration::from_millis(100)).await; @@ -134,7 +134,7 @@ let v1_cid = Cid::from_str(v1_cid_str).unwrap(); let repos = get_test_repos().await; - let cursor = repos.repo.get_max_seq().await.unwrap().as_i64(); + let cursor = flushed_max_seq(repos).await.as_i64(); let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await; tokio::time::sleep(std::time::Duration::from_millis(100)).await; @@ -203,7 +203,7 @@ let rkey = parts[parts.len() - 1]; let repos = get_test_repos().await; - let cursor = repos.repo.get_max_seq().await.unwrap().as_i64(); + let cursor = flushed_max_seq(repos).await.as_i64(); let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await; tokio::time::sleep(std::time::Duration::from_millis(100)).await; @@ -246,7 +246,7 @@ let (token, did) = create_account_and_login(&client).await; let repos = get_test_repos().await; - let cursor = repos.repo.get_max_seq().await.unwrap().as_i64(); + let cursor = flushed_max_seq(repos).await.as_i64(); let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await; tokio::time::sleep(std::time::Duration::from_millis(100)).await; @@ -315,7 +315,7 @@ let (token, did) = create_account_and_login(&client).await; let repos = get_test_repos().await; - let cursor = repos.repo.get_max_seq().await.unwrap().as_i64(); + let cursor = flushed_max_seq(repos).await.as_i64(); let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await; tokio::time::sleep(std::time::Duration::from_millis(100)).await; @@ -396,7 +396,7 @@ }; let repos = get_test_repos().await; - let cursor = repos.repo.get_max_seq().await.unwrap().as_i64(); + let cursor = flushed_max_seq(repos).await.as_i64(); let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await; tokio::time::sleep(std::time::Duration::from_millis(100)).await; @@ -444,7 +444,7 @@ let (token, did) = create_account_and_login(&client).await; let repos = get_test_repos().await; - let baseline_seq = repos.repo.get_max_seq().await.unwrap().as_i64(); + let baseline_seq = flushed_max_seq(repos).await.as_i64(); let mut expected_cids: Vec = Vec::with_capacity(5); let texts = [ @@ -496,7 +496,7 @@ let (bob_token, bob_did) = create_account_and_login(&client).await; let repos = get_test_repos().await; - let cursor = repos.repo.get_max_seq().await.unwrap().as_i64(); + let cursor = flushed_max_seq(repos).await.as_i64(); let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await; tokio::time::sleep(std::time::Duration::from_millis(100)).await; diff --git a/crates/tranquil-pds/tests/store_parity.rs b/crates/tranquil-pds/tests/store_parity.rs --- a/crates/tranquil-pds/tests/store_parity.rs +++ b/crates/tranquil-pds/tests/store_parity.rs @@ -1628,10 +1628,12 @@ rev: Some("rev0".to_string()), }; - let pg_seq = f.pg.repo.insert_commit_event(&event).await.unwrap(); - let store_seq = f.store.repo.insert_commit_event(&event).await.unwrap(); - assert!(pg_seq.as_i64() > 0); - assert!(store_seq.as_i64() > 0); + let baseline = f.pg.repo.get_max_seq().await.unwrap(); + f.pg.repo.insert_commit_event(&event).await.unwrap(); + f.store.repo.insert_commit_event(&event).await.unwrap(); + let pg_seq = common::sequenced_event_for_did(&f.pg, baseline, &did) + .await + .seq; let past_cutoff = chrono::Utc::now() - chrono::Duration::hours(24); let pg_pruned_past = diff --git a/crates/tranquil-store/tests/sim_metastore.rs b/crates/tranquil-store/tests/sim_metastore.rs --- a/crates/tranquil-store/tests/sim_metastore.rs +++ b/crates/tranquil-store/tests/sim_metastore.rs @@ -405,13 +405,9 @@ ), }; - let result = h.apply_commit(input).unwrap_or_else(|e| { + h.apply_commit(input).unwrap_or_else(|e| { panic!("seed={seed} commit {commit_idx}: {e:?}") }); - assert!( - result.seq > 0, - "seed={seed} commit {commit_idx} seq must be positive" - ); let new_persisted = match commit_idx <= crash_after { true => { @@ -689,12 +685,7 @@ receivers.into_iter().for_each(|(idx, rx)| { let result = rt.block_on(rx); match result { - Ok(Ok(commit_result)) => { - assert!( - commit_result.seq > 0, - "seed={seed} idx={idx} commit seq must be positive" - ); - } + Ok(Ok(_commit_result)) => {} Ok(Err(e)) => { panic!("seed={seed} idx={idx} commit failed: {e:?}"); } diff --git a/crates/tranquil-sync/src/listener.rs b/crates/tranquil-sync/src/listener.rs --- a/crates/tranquil-sync/src/listener.rs +++ b/crates/tranquil-sync/src/listener.rs @@ -1,10 +1,14 @@ use std::sync::atomic::{AtomicI64, Ordering}; -use tracing::{debug, error, info, warn}; +use std::time::Duration; +use tracing::{error, info, warn}; use tranquil_db_traits::SequenceNumber; use tranquil_pds::state::AppState; use tranquil_pds::sync::firehose::SequencedEvent; static LAST_BROADCAST_SEQ: AtomicI64 = AtomicI64::new(0); + +const DRAIN_BATCH_SIZE: i64 = 1000; +const POLL_INTERVAL: Duration = Duration::from_secs(1); pub async fn start_sequencer_listener(state: AppState) { let initial_seq = state @@ -23,7 +27,7 @@ loop { if let Err(e) = listen_loop(state.clone()).await { error!("Sequencer listener failed: {}. Restarting in 5s...", e); - tokio::time::sleep(tokio::time::Duration::from_secs(5)).await; + tokio::time::sleep(Duration::from_secs(5)).await; } } }); @@ -37,88 +41,53 @@ .await .map_err(|e| anyhow::anyhow!("Failed to subscribe to events: {:?}", e))?; info!("Connected to database and listening for repo updates"); - let catchup_start = SequenceNumber::from_raw(LAST_BROADCAST_SEQ.load(Ordering::SeqCst)); - let events = state - .repos - .repo - .get_events_since_seq(catchup_start, None) - .await - .map_err(|e| anyhow::anyhow!("Failed to fetch catchup events: {:?}", e))?; - if !events.is_empty() { - info!( - count = events.len(), - from_seq = catchup_start.as_i64(), - "Broadcasting catch-up events" - ); - events.into_iter().for_each(|event| { - let seq = event.seq; - let firehose_event = to_firehose_event(event); - let _ = state.firehose_tx.send(firehose_event); - LAST_BROADCAST_SEQ.store(seq.as_i64(), Ordering::SeqCst); - }); + + let mut last_seq = LAST_BROADCAST_SEQ.load(Ordering::SeqCst); + sequence_and_broadcast(&state, &mut last_seq).await; + + loop { + tokio::select! { + received = receiver.recv() => { + if received.is_none() { + return Err(anyhow::anyhow!("Event receiver disconnected")); + } + } + _ = tokio::time::sleep(POLL_INTERVAL) => {} + } + sequence_and_broadcast(&state, &mut last_seq).await; + } +} + +async fn sequence_and_broadcast(state: &AppState, last_seq: &mut i64) { + if let Err(e) = state.repos.repo.assign_pending_sequences().await { + warn!("Failed to assign pending firehose sequences: {:?}", e); } loop { - let Some(seq_id) = receiver.recv().await else { - return Err(anyhow::anyhow!("Event receiver disconnected")); - }; - debug!(seq = seq_id, "Received event notification"); - let last_seq = LAST_BROADCAST_SEQ.load(Ordering::SeqCst); - if seq_id <= last_seq { - debug!( - seq = seq_id, - last = last_seq, - "Skipping already-broadcast event" - ); - continue; - } - if seq_id > last_seq + 1 { - let gap_events = state - .repos - .repo - .get_events_in_seq_range( - SequenceNumber::from_raw(last_seq), - SequenceNumber::from_raw(seq_id), - ) - .await - .unwrap_or_default(); - if !gap_events.is_empty() { - debug!(count = gap_events.len(), "Filling sequence gap"); - gap_events.into_iter().for_each(|event| { - let seq = event.seq; - let firehose_event = to_firehose_event(event); - let _ = state.firehose_tx.send(firehose_event); - LAST_BROADCAST_SEQ.store(seq.as_i64(), Ordering::SeqCst); - }); - } - } - let event = state + let events = match state .repos .repo - .get_event_by_seq(SequenceNumber::from_raw(seq_id)) + .get_events_since_seq(SequenceNumber::from_raw(*last_seq), Some(DRAIN_BATCH_SIZE)) .await - .ok() - .flatten(); - if let Some(event) = event { - let seq = event.seq; - let firehose_event = to_firehose_event(event); - match state.firehose_tx.send(firehose_event) { - Ok(receiver_count) => { - debug!( - seq = seq_id, - receivers = receiver_count, - "Broadcast event to firehose" - ); - } - Err(e) => { - warn!(seq = seq_id, error = %e, "Failed to broadcast event (no receivers?)"); - } + { + Ok(events) => events, + Err(e) => { + warn!("Sequencer broadcast query failed: {:?}", e); + return; } - LAST_BROADCAST_SEQ.store(seq.as_i64(), Ordering::SeqCst); - } else { - warn!( - seq = seq_id, - "Received notification but could not find row in repo_seq" - ); + }; + if events.is_empty() { + return; + } + let batch_len = events.len(); + for event in events { + let seq = event.seq.as_i64(); + let firehose_event = to_firehose_event(event); + let _ = state.firehose_tx.send(firehose_event); + *last_seq = seq; + LAST_BROADCAST_SEQ.store(seq, Ordering::SeqCst); + } + if (batch_len as i64) < DRAIN_BATCH_SIZE { + return; } } } diff --git a/crates/tranquil-api/src/repo/import.rs b/crates/tranquil-api/src/repo/import.rs --- a/crates/tranquil-api/src/repo/import.rs +++ b/crates/tranquil-api/src/repo/import.rs @@ -417,7 +417,6 @@ rev: None, }; - let seq = state.repos.repo.insert_commit_event(&data).await?; - state.repos.repo.notify_update(seq).await?; + state.repos.repo.insert_commit_event(&data).await?; Ok(()) } diff --git a/crates/tranquil-api/src/server/account_status.rs b/crates/tranquil-api/src/server/account_status.rs --- a/crates/tranquil-api/src/server/account_status.rs +++ b/crates/tranquil-api/src/server/account_status.rs @@ -665,27 +665,22 @@ error!("DB error deleting account: {:?}", e); ApiError::InternalError(None) })?; - let account_seq = tranquil_pds::repo_ops::sequence_account_event( + if let Err(e) = tranquil_pds::repo_ops::sequence_account_event( &state, did, tranquil_db_traits::AccountStatus::Deleted, ) - .await; - match account_seq { - Ok(seq) => { - if let Err(e) = state.repos.repo.delete_sequences_except(did, seq).await { - warn!( - "Failed to cleanup sequences for deleted account {}: {}", - did, e - ); - } - } - Err(e) => { - warn!( - "Failed to sequence account deletion event for {}: {}", - did, e - ); - } + .await + { + warn!( + "Failed to sequence account deletion event for {}: {}", + did, e + ); + } else if let Err(e) = state.repos.repo.purge_did_events_keeping_latest(did).await { + warn!( + "Failed to cleanup sequences for deleted account {}: {}", + did, e + ); } let _ = state .cache diff --git a/crates/tranquil-db/src/postgres/event_notifier.rs b/crates/tranquil-db/src/postgres/event_notifier.rs --- a/crates/tranquil-db/src/postgres/event_notifier.rs +++ b/crates/tranquil-db/src/postgres/event_notifier.rs @@ -35,9 +35,9 @@ #[async_trait] impl RepoEventReceiver for PostgresRepoEventReceiver { - async fn recv(&mut self) -> Option { + async fn recv(&mut self) -> Option<()> { match self.listener.recv().await { - Ok(notification) => notification.payload().parse().ok(), + Ok(_) => Some(()), Err(_) => None, } } diff --git a/crates/tranquil-db/src/postgres/repo.rs b/crates/tranquil-db/src/postgres/repo.rs --- a/crates/tranquil-db/src/postgres/repo.rs +++ b/crates/tranquil-db/src/postgres/repo.rs @@ -111,6 +111,64 @@ }) } +const SEQUENCER_LOCK_KEY: i64 = 0x0074_7261_6e73_6571; +const SEQUENCER_BATCH_SIZE: i64 = 1000; + +async fn notify_repo_pending(pool: &PgPool) { + if let Err(e) = sqlx::query!("NOTIFY repo_updates").execute(pool).await { + tracing::warn!(error = %e, "failed to notify repo_updates listeners"); + } +} + +async fn assign_one_batch( + mut tx: sqlx::Transaction<'_, sqlx::Postgres>, + pool: &PgPool, +) -> Result { + let pending_ids: Vec = sqlx::query_scalar!( + r#"SELECT id as "id!" FROM repo_seq WHERE seq IS NULL ORDER BY id LIMIT $1"#, + SEQUENCER_BATCH_SIZE + ) + .fetch_all(&mut *tx) + .await + .map_err(map_sqlx_error)?; + + let count = pending_ids.len() as i64; + if count == 0 { + tx.commit().await.map_err(map_sqlx_error)?; + return Ok(0); + } + + let start = sqlx::query_scalar!(r#"SELECT nextval('firehose_seq') as "seq!""#) + .fetch_one(&mut *tx) + .await + .map_err(map_sqlx_error)?; + if count > 1 { + sqlx::query!("SELECT setval('firehose_seq', $1)", start + count - 1) + .fetch_one(&mut *tx) + .await + .map_err(map_sqlx_error)?; + } + + let assigned = sqlx::query!( + r#"UPDATE repo_seq r + SET seq = $1::bigint + (o.ord - 1) + FROM unnest($2::bigint[]) WITH ORDINALITY AS o(id, ord) + WHERE r.id = o.id"#, + start, + &pending_ids + ) + .execute(&mut *tx) + .await + .map_err(map_sqlx_error)? + .rows_affected(); + + tx.commit().await.map_err(map_sqlx_error)?; + if assigned > 0 { + notify_repo_pending(pool).await; + } + Ok(count) +} + pub struct PostgresRepoRepository { pool: PgPool, } @@ -716,13 +774,12 @@ Ok(rows.into_iter().map(|(cid,)| cid).collect()) } - async fn insert_commit_event(&self, data: &CommitEventData) -> Result { + async fn insert_commit_event(&self, data: &CommitEventData) -> Result<(), DbError> { let (block_cids, block_data) = inline_to_paired_blocks(data.blocks.as_deref()); - let seq = sqlx::query_scalar!( + sqlx::query!( r#" INSERT INTO repo_seq (did, event_type, commit_cid, prev_cid, ops, blobs, block_cids, block_data, prev_data_cid, rev) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) - RETURNING seq "#, data.did.as_str(), data.event_type.as_str(), @@ -735,67 +792,54 @@ data.prev_data_cid.as_ref().map(|c| c.as_str()), data.rev ) - .fetch_one(&self.pool) + .execute(&self.pool) .await .map_err(map_sqlx_error)?; - Ok(seq.into()) + notify_repo_pending(&self.pool).await; + Ok(()) } async fn insert_identity_event( &self, did: &Did, handle: Option<&Handle>, - ) -> Result { + ) -> Result<(), DbError> { let handle_str = handle.map(|h| h.as_str()); - let seq = sqlx::query_scalar!( + sqlx::query!( r#" INSERT INTO repo_seq (did, event_type, handle) VALUES ($1, 'identity', $2) - RETURNING seq "#, did.as_str(), handle_str ) - .fetch_one(&self.pool) + .execute(&self.pool) .await .map_err(map_sqlx_error)?; - sqlx::query(&format!("NOTIFY repo_updates, '{}'", seq)) - .execute(&self.pool) - .await - .map_err(map_sqlx_error)?; - - Ok(seq.into()) + notify_repo_pending(&self.pool).await; + Ok(()) } - async fn insert_account_event( - &self, - did: &Did, - status: AccountStatus, - ) -> Result { + async fn insert_account_event(&self, did: &Did, status: AccountStatus) -> Result<(), DbError> { let active = status.is_active(); let status_str = status.for_firehose(); - let seq = sqlx::query_scalar!( + sqlx::query!( r#" INSERT INTO repo_seq (did, event_type, active, status) VALUES ($1, 'account', $2, $3) - RETURNING seq "#, did.as_str(), active, status_str ) - .fetch_one(&self.pool) + .execute(&self.pool) .await .map_err(map_sqlx_error)?; - sqlx::query(&format!("NOTIFY repo_updates, '{}'", seq)) - .execute(&self.pool) - .await - .map_err(map_sqlx_error)?; - - Ok(seq.into()) + notify_repo_pending(&self.pool).await; + Ok(()) } async fn insert_sync_event( @@ -804,18 +848,17 @@ commit_cid: &CidLink, rev: Option<&str>, commit_bytes: &[u8], - ) -> Result { + ) -> Result<(), DbError> { let cid_bytes = commit_cid .to_cid() .map(|c| c.to_bytes()) .unwrap_or_default(); let block_cids: Vec> = vec![cid_bytes]; let block_data: Vec> = vec![commit_bytes.to_vec()]; - let seq = sqlx::query_scalar!( + sqlx::query!( r#" INSERT INTO repo_seq (did, event_type, commit_cid, rev, block_cids, block_data) VALUES ($1, 'sync', $2, $3, $4, $5) - RETURNING seq "#, did.as_str(), commit_cid.as_str(), @@ -823,16 +866,12 @@ &block_cids as &[Vec], &block_data as &[Vec] ) - .fetch_one(&self.pool) + .execute(&self.pool) .await .map_err(map_sqlx_error)?; - sqlx::query(&format!("NOTIFY repo_updates, '{}'", seq)) - .execute(&self.pool) - .await - .map_err(map_sqlx_error)?; - - Ok(seq.into()) + notify_repo_pending(&self.pool).await; + Ok(()) } async fn insert_genesis_commit_event( @@ -843,7 +882,7 @@ rev: &str, commit_bytes: &[u8], mst_root_bytes: &[u8], - ) -> Result { + ) -> Result<(), DbError> { let ops = serde_json::json!([]); let blobs: Vec = vec![]; let commit_cid_bytes = commit_cid @@ -858,11 +897,10 @@ let block_data: Vec> = vec![commit_bytes.to_vec(), mst_root_bytes.to_vec()]; let prev_cid: Option<&str> = None; - let seq = sqlx::query_scalar!( + sqlx::query!( r#" INSERT INTO repo_seq (did, event_type, commit_cid, prev_cid, ops, blobs, block_cids, block_data, rev) VALUES ($1, 'commit', $2, $3::TEXT, $4, $5, $6, $7, $8) - RETURNING seq "#, did.as_str(), commit_cid.as_str(), @@ -873,33 +911,64 @@ &block_data as &[Vec], rev ) - .fetch_one(&self.pool) + .execute(&self.pool) .await .map_err(map_sqlx_error)?; - sqlx::query(&format!("NOTIFY repo_updates, '{}'", seq)) - .execute(&self.pool) - .await - .map_err(map_sqlx_error)?; - - Ok(seq.into()) + notify_repo_pending(&self.pool).await; + Ok(()) } - async fn delete_sequences_except( - &self, - did: &Did, - keep_seq: SequenceNumber, - ) -> Result<(), DbError> { + async fn purge_did_events_keeping_latest(&self, did: &Did) -> Result<(), DbError> { sqlx::query!( - "DELETE FROM repo_seq WHERE did = $1 AND seq != $2", - did.as_str(), - keep_seq.as_i64() + r#" + DELETE FROM repo_seq + WHERE did = $1 + AND id <> (SELECT id FROM repo_seq WHERE did = $1 ORDER BY id DESC LIMIT 1) + "#, + did.as_str() ) .execute(&self.pool) .await .map_err(map_sqlx_error)?; Ok(()) + } + + async fn assign_pending_sequences(&self) -> Result { + let mut total: u64 = 0; + loop { + let mut tx = self.pool.begin().await.map_err(map_sqlx_error)?; + let acquired = sqlx::query_scalar!( + r#"SELECT pg_try_advisory_xact_lock($1) as "locked!""#, + SEQUENCER_LOCK_KEY + ) + .fetch_one(&mut *tx) + .await + .map_err(map_sqlx_error)?; + if !acquired { + return Ok(total); + } + let count = assign_one_batch(tx, &self.pool).await?; + total += count as u64; + if count < SEQUENCER_BATCH_SIZE { + return Ok(total); + } + } + } + + async fn flush_pending_sequences(&self) -> Result<(), DbError> { + loop { + let mut tx = self.pool.begin().await.map_err(map_sqlx_error)?; + sqlx::query!("SELECT pg_advisory_xact_lock($1)", SEQUENCER_LOCK_KEY) + .execute(&mut *tx) + .await + .map_err(map_sqlx_error)?; + let count = assign_one_batch(tx, &self.pool).await?; + if count < SEQUENCER_BATCH_SIZE { + return Ok(()); + } + } } async fn prune_events_older_than(&self, cutoff: DateTime) -> Result { @@ -965,7 +1034,7 @@ Some(lim) => { let rows = sqlx::query_as!( SequencedEventRow, - r#"SELECT seq, did, created_at, event_type as "event_type: RepoEventType", commit_cid, prev_cid, prev_data_cid, + r#"SELECT seq as "seq!", did, created_at, event_type as "event_type: RepoEventType", commit_cid, prev_cid, prev_data_cid, ops, blobs, block_cids, block_data, blocks_cids, handle, active, status, rev FROM repo_seq WHERE seq > $1 @@ -982,7 +1051,7 @@ None => { let rows = sqlx::query_as!( SequencedEventRow, - r#"SELECT seq, did, created_at, event_type as "event_type: RepoEventType", commit_cid, prev_cid, prev_data_cid, + r#"SELECT seq as "seq!", did, created_at, event_type as "event_type: RepoEventType", commit_cid, prev_cid, prev_data_cid, ops, blobs, block_cids, block_data, blocks_cids, handle, active, status, rev FROM repo_seq WHERE seq > $1 @@ -1004,7 +1073,7 @@ ) -> Result, DbError> { let rows = sqlx::query_as!( SequencedEventRow, - r#"SELECT seq, did, created_at, event_type as "event_type: RepoEventType", commit_cid, prev_cid, prev_data_cid, + r#"SELECT seq as "seq!", did, created_at, event_type as "event_type: RepoEventType", commit_cid, prev_cid, prev_data_cid, ops, blobs, block_cids, block_data, blocks_cids, handle, active, status, rev FROM repo_seq WHERE seq > $1 AND seq < $2 @@ -1024,7 +1093,7 @@ ) -> Result, DbError> { let row = sqlx::query_as!( SequencedEventRow, - r#"SELECT seq, did, created_at, event_type as "event_type: RepoEventType", commit_cid, prev_cid, prev_data_cid, + r#"SELECT seq as "seq!", did, created_at, event_type as "event_type: RepoEventType", commit_cid, prev_cid, prev_data_cid, ops, blobs, block_cids, block_data, blocks_cids, handle, active, status, rev FROM repo_seq WHERE seq = $1"#, @@ -1043,7 +1112,7 @@ ) -> Result, DbError> { let rows = sqlx::query_as!( SequencedEventRow, - r#"SELECT seq, did, created_at, event_type as "event_type: RepoEventType", commit_cid, prev_cid, prev_data_cid, + r#"SELECT seq as "seq!", did, created_at, event_type as "event_type: RepoEventType", commit_cid, prev_cid, prev_data_cid, ops, blobs, block_cids, block_data, blocks_cids, handle, active, status, rev FROM repo_seq WHERE seq > $1 @@ -1102,14 +1171,6 @@ .await .map_err(map_sqlx_error)?; Ok(cid.map(CidLink::from)) - } - - async fn notify_update(&self, seq: SequenceNumber) -> Result<(), DbError> { - sqlx::query(&format!("NOTIFY repo_updates, '{}'", seq.as_i64())) - .execute(&self.pool) - .await - .map_err(map_sqlx_error)?; - Ok(()) } async fn import_repo_data( @@ -1402,28 +1463,27 @@ let event = input.commit_event; let (event_block_cids, event_block_data) = inline_into_paired_blocks(event.blocks); - let seq: i64 = sqlx::query_scalar( + sqlx::query!( r#" INSERT INTO repo_seq (did, event_type, commit_cid, prev_cid, ops, blobs, block_cids, block_data, prev_data_cid, rev) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) - RETURNING seq "#, + event.did.as_str(), + event.event_type.as_str(), + event.commit_cid.as_ref().map(|c| c.as_str()), + event.prev_cid.as_ref().map(|c| c.as_str()), + event.ops, + event.blobs.as_deref(), + &event_block_cids as &[Vec], + &event_block_data as &[Vec], + event.prev_data_cid.as_ref().map(|c| c.as_str()), + event.rev ) - .bind(event.did.as_str()) - .bind(event.event_type.as_str()) - .bind(event.commit_cid.as_ref().map(|c| c.as_str())) - .bind(event.prev_cid.as_ref().map(|c| c.as_str())) - .bind(&event.ops) - .bind(&event.blobs) - .bind(&event_block_cids) - .bind(&event_block_data) - .bind(event.prev_data_cid.as_ref().map(|c| c.as_str())) - .bind(&event.rev) - .fetch_one(&mut *tx) + .execute(&mut *tx) .await .map_err(|e| ApplyCommitError::Database(e.to_string()))?; - sqlx::query(&format!("NOTIFY repo_updates, '{}'", seq)) + sqlx::query!("NOTIFY repo_updates") .execute(&mut *tx) .await .map_err(|e| ApplyCommitError::Database(e.to_string()))?; @@ -1432,10 +1492,7 @@ .await .map_err(|e| ApplyCommitError::Database(e.to_string()))?; - Ok(tranquil_db_traits::ApplyCommitResult { - seq, - is_account_active, - }) + Ok(tranquil_db_traits::ApplyCommitResult { is_account_active }) } async fn get_users_without_blocks(&self) -> Result, DbError> { diff --git a/crates/tranquil-db/src/postgres/user.rs b/crates/tranquil-db/src/postgres/user.rs --- a/crates/tranquil-db/src/postgres/user.rs +++ b/crates/tranquil-db/src/postgres/user.rs @@ -2291,7 +2291,7 @@ .collect()) } - async fn delete_account_with_firehose(&self, user_id: Uuid, did: &Did) -> Result { + async fn delete_account_with_firehose(&self, user_id: Uuid, did: &Did) -> Result<(), DbError> { let mut tx = self.pool.begin().await.map_err(map_sqlx_error)?; sqlx::query!("DELETE FROM blobs WHERE created_by_user = $1", user_id) @@ -2370,11 +2370,11 @@ .await .map_err(map_sqlx_error)?; - let account_seq: i64 = sqlx::query_scalar!( + let event_id: i64 = sqlx::query_scalar!( r#" INSERT INTO repo_seq (did, event_type, active, status) VALUES ($1, 'account', false, 'deleted') - RETURNING seq + RETURNING id "#, did.as_str() ) @@ -2383,9 +2383,9 @@ .map_err(map_sqlx_error)?; sqlx::query!( - "DELETE FROM repo_seq WHERE did = $1 AND seq != $2", + "DELETE FROM repo_seq WHERE did = $1 AND id <> $2", did.as_str(), - account_seq + event_id ) .execute(&mut *tx) .await @@ -2393,12 +2393,12 @@ tx.commit().await.map_err(map_sqlx_error)?; - sqlx::query(&format!("NOTIFY repo_updates, '{}'", account_seq)) + sqlx::query!("NOTIFY repo_updates") .execute(&self.pool) .await .map_err(map_sqlx_error)?; - Ok(account_seq) + Ok(()) } async fn create_password_account( diff --git a/crates/tranquil-pds/tests/common/mod.rs b/crates/tranquil-pds/tests/common/mod.rs --- a/crates/tranquil-pds/tests/common/mod.rs +++ b/crates/tranquil-pds/tests/common/mod.rs @@ -921,6 +921,40 @@ .expect("TEST_BLOCK_STORE not initialized") } +#[allow(dead_code)] +pub async fn flushed_max_seq( + repos: &tranquil_db::PostgresRepositories, +) -> tranquil_db_traits::SequenceNumber { + repos + .repo + .flush_pending_sequences() + .await + .expect("flush_pending_sequences"); + repos.repo.get_max_seq().await.expect("get_max_seq") +} + +#[allow(dead_code)] +pub async fn sequenced_event_for_did( + repos: &tranquil_db::PostgresRepositories, + baseline: tranquil_db_traits::SequenceNumber, + did: &tranquil_types::Did, +) -> tranquil_db_traits::SequencedEvent { + repos + .repo + .flush_pending_sequences() + .await + .expect("flush_pending_sequences"); + repos + .repo + .get_events_since_seq(baseline, None) + .await + .expect("get_events_since_seq") + .into_iter() + .filter(|event| &event.did == did) + .last() + .unwrap_or_else(|| panic!("event for did {did} not found after flush")) +} + fn extract_verification_code(body_text: &str) -> String { let lines: Vec<&str> = body_text.lines().collect(); lines diff --git a/crates/tranquil-store/src/eventlog/notifier.rs b/crates/tranquil-store/src/eventlog/notifier.rs --- a/crates/tranquil-store/src/eventlog/notifier.rs +++ b/crates/tranquil-store/src/eventlog/notifier.rs @@ -30,7 +30,7 @@ #[async_trait] impl RepoEventReceiver for EventLogEventReceiver { - async fn recv(&mut self) -> Option { - self.subscriber.next().await.map(|event| event.seq.as_i64()) + async fn recv(&mut self) -> Option<()> { + self.subscriber.next().await.map(|_| ()) } } diff --git a/crates/tranquil-store/src/metastore/client.rs b/crates/tranquil-store/src/metastore/client.rs --- a/crates/tranquil-store/src/metastore/client.rs +++ b/crates/tranquil-store/src/metastore/client.rs @@ -474,21 +474,21 @@ recv(rx).await } - async fn insert_commit_event(&self, data: &CommitEventData) -> Result { + async fn insert_commit_event(&self, data: &CommitEventData) -> Result<(), DbError> { let (tx, rx) = oneshot::channel(); self.pool .send(MetastoreRequest::Event(EventRequest::InsertCommitEvent { data: data.clone(), tx, }))?; - recv(rx).await + recv(rx).await.map(|_: SequenceNumber| ()) } async fn insert_identity_event( &self, did: &Did, handle: Option<&Handle>, - ) -> Result { + ) -> Result<(), DbError> { let (tx, rx) = oneshot::channel(); self.pool .send(MetastoreRequest::Event(EventRequest::InsertIdentityEvent { @@ -496,14 +496,10 @@ handle: handle.cloned(), tx, }))?; - recv(rx).await + recv(rx).await.map(|_: SequenceNumber| ()) } - async fn insert_account_event( - &self, - did: &Did, - status: AccountStatus, - ) -> Result { + async fn insert_account_event(&self, did: &Did, status: AccountStatus) -> Result<(), DbError> { let (tx, rx) = oneshot::channel(); self.pool .send(MetastoreRequest::Event(EventRequest::InsertAccountEvent { @@ -511,7 +507,7 @@ status, tx, }))?; - recv(rx).await + recv(rx).await.map(|_: SequenceNumber| ()) } async fn insert_sync_event( @@ -520,7 +516,7 @@ commit_cid: &CidLink, rev: Option<&str>, commit_bytes: &[u8], - ) -> Result { + ) -> Result<(), DbError> { let (tx, rx) = oneshot::channel(); self.pool .send(MetastoreRequest::Event(EventRequest::InsertSyncEvent { @@ -530,7 +526,7 @@ commit_bytes: commit_bytes.to_vec(), tx, }))?; - recv(rx).await + recv(rx).await.map(|_: SequenceNumber| ()) } async fn insert_genesis_commit_event( @@ -541,7 +537,7 @@ rev: &str, commit_bytes: &[u8], mst_root_bytes: &[u8], - ) -> Result { + ) -> Result<(), DbError> { let (tx, rx) = oneshot::channel(); self.pool.send(MetastoreRequest::Event( EventRequest::InsertGenesisCommitEvent { @@ -554,19 +550,14 @@ tx, }, ))?; - recv(rx).await + recv(rx).await.map(|_: SequenceNumber| ()) } - async fn delete_sequences_except( - &self, - did: &Did, - keep_seq: SequenceNumber, - ) -> Result<(), DbError> { + async fn purge_did_events_keeping_latest(&self, did: &Did) -> Result<(), DbError> { let (tx, rx) = oneshot::channel(); self.pool.send(MetastoreRequest::Event( - EventRequest::DeleteSequencesExcept { + EventRequest::PurgeDidEventsKeepingLatest { did: did.clone(), - keep_seq, tx, }, ))?; @@ -710,16 +701,6 @@ self.pool.send(MetastoreRequest::Repo( RepoRequest::GetRepoRootCidByUserId { user_id, tx }, ))?; - recv(rx).await - } - - async fn notify_update(&self, seq: SequenceNumber) -> Result<(), DbError> { - let (tx, rx) = oneshot::channel(); - self.pool - .send(MetastoreRequest::Event(EventRequest::NotifyUpdate { - seq, - tx, - }))?; recv(rx).await } @@ -4898,7 +4879,7 @@ recv(rx).await } - async fn delete_account_with_firehose(&self, user_id: Uuid, did: &Did) -> Result { + async fn delete_account_with_firehose(&self, user_id: Uuid, did: &Did) -> Result<(), DbError> { let (tx, rx) = oneshot::channel(); self.pool.send(MetastoreRequest::User( UserRequest::DeleteAccountWithFirehose { diff --git a/crates/tranquil-store/src/metastore/commit_ops.rs b/crates/tranquil-store/src/metastore/commit_ops.rs --- a/crates/tranquil-store/src/metastore/commit_ops.rs +++ b/crates/tranquil-store/src/metastore/commit_ops.rs @@ -245,7 +245,7 @@ .serialize() .map_err(|e| ApplyCommitError::Database(e.to_string()))?; - let (seq, deferred) = self + let (_seq, deferred) = self .event_ops .append_commit_event_into_batch( &mut batch, @@ -261,7 +261,6 @@ self.event_ops.complete_broadcast(deferred); Ok(ApplyCommitResult { - seq: seq.as_i64(), is_account_active: is_active, }) } @@ -477,7 +476,7 @@ use crate::eventlog::{EventLog, EventLogConfig}; use crate::io::RealIO; use crate::metastore::{Metastore, MetastoreConfig}; - use tranquil_db_traits::{CommitEventData, RepoEventType, SequenceNumber}; + use tranquil_db_traits::{CommitEventData, RepoEventType}; use tranquil_types::{Handle, Nsid, Rkey}; struct TestHarness { @@ -599,7 +598,6 @@ }; let result = ops.apply_commit(input).unwrap(); - assert!(result.seq > 0); assert!(result.is_account_active); let repo = h.metastore.repo_ops().get_repo(user_id).unwrap().unwrap(); @@ -812,8 +810,8 @@ }, }; - let result = ops.apply_commit(input).unwrap(); - let seq = SequenceNumber::from_raw(result.seq); + ops.apply_commit(input).unwrap(); + let seq = ops.event_ops.get_max_seq(); let event = ops.event_ops.get_event_by_seq(seq).unwrap().unwrap(); assert_eq!(event.did, did); @@ -954,8 +952,7 @@ }, }; - let result = ops.apply_commit(input).unwrap(); - assert!(result.seq > 0); + ops.apply_commit(input).unwrap(); } #[test] @@ -1144,8 +1141,7 @@ }, }; - let result = ops.apply_commit(input).unwrap(); - assert!(result.seq > 0); + ops.apply_commit(input).unwrap(); metastore.persist().unwrap(); } diff --git a/crates/tranquil-store/src/metastore/event_ops.rs b/crates/tranquil-store/src/metastore/event_ops.rs --- a/crates/tranquil-store/src/metastore/event_ops.rs +++ b/crates/tranquil-store/src/metastore/event_ops.rs @@ -41,10 +41,6 @@ self.bridge.notifier() } - pub fn notify_update(&self, _seq: SequenceNumber) -> Result<(), DbError> { - Ok(()) - } - pub fn insert_commit_event(&self, data: &CommitEventData) -> Result { let event = Self::build_commit_event(data); self.append_and_index(&event, &data.did, data.rev.as_deref()) @@ -380,6 +376,33 @@ batch.commit().map_err(fjall_to_db)?; Ok(()) + } + + pub fn purge_did_events_keeping_latest(&self, did: &Did) -> Result<(), DbError> { + let user_hash = UserHash::from_did(did.as_str()); + let prefix = did_events_prefix(user_hash); + let upper = exclusive_upper_bound(prefix.as_slice()) + .expect("did_events prefix can never be all-0xFF"); + + let latest = self + .repo_data + .range(prefix.as_slice()..upper.as_slice()) + .map(|guard| { + let (key, _) = guard.into_inner().map_err(fjall_to_db)?; + decode_did_events_seq(key.as_ref()) + }) + .collect::, DbError>>()? + .into_iter() + .max(); + + match latest { + Some(seq) => { + let keep = i64::try_from(seq) + .map_err(|_| DbError::Query("sequence number out of range".to_owned()))?; + self.delete_sequences_except(did, SequenceNumber::from_raw(keep)) + } + None => Ok(()), + } } pub fn read_last_applied_cursor(&self) -> Result, DbError> { diff --git a/crates/tranquil-store/src/metastore/handler.rs b/crates/tranquil-store/src/metastore/handler.rs --- a/crates/tranquil-store/src/metastore/handler.rs +++ b/crates/tranquil-store/src/metastore/handler.rs @@ -362,9 +362,8 @@ mst_root_bytes: Vec, tx: Tx, }, - DeleteSequencesExcept { + PurgeDidEventsKeepingLatest { did: Did, - keep_seq: SequenceNumber, tx: Tx<()>, }, GetMaxSeq { @@ -393,10 +392,6 @@ limit: i64, tx: Tx>, }, - NotifyUpdate { - seq: SequenceNumber, - tx: Tx<()>, - }, } impl EventRequest { @@ -409,7 +404,7 @@ | Self::InsertAccountEvent { did, .. } | Self::InsertSyncEvent { did, .. } | Self::InsertGenesisCommitEvent { did, .. } - | Self::DeleteSequencesExcept { did, .. } => { + | Self::PurgeDidEventsKeepingLatest { did, .. } => { Routing::Sharded(UserHash::from_did(did.as_str()).raw()) } Self::GetMaxSeq { .. } @@ -417,8 +412,7 @@ | Self::GetEventsSinceSeq { .. } | Self::GetEventsInSeqRange { .. } | Self::GetEventBySeq { .. } - | Self::GetEventsSinceCursor { .. } - | Self::NotifyUpdate { .. } => Routing::Global, + | Self::GetEventsSinceCursor { .. } => Routing::Global, } } } @@ -1513,7 +1507,7 @@ DeleteAccountWithFirehose { user_id: Uuid, did: Did, - tx: Tx, + tx: Tx<()>, }, CreatePasswordAccount { input: CreatePasswordAccountInput, @@ -2967,8 +2961,8 @@ ); let _ = tx.send(result); } - EventRequest::DeleteSequencesExcept { did, keep_seq, tx } => { - let result = state.event_ops.delete_sequences_except(&did, keep_seq); + EventRequest::PurgeDidEventsKeepingLatest { did, tx } => { + let result = state.event_ops.purge_did_events_keeping_latest(&did); let _ = tx.send(result); } EventRequest::GetMaxSeq { tx } => { @@ -2996,9 +2990,6 @@ } EventRequest::GetEventsSinceCursor { cursor, limit, tx } => { let _ = tx.send(state.event_ops.get_events_since_cursor(cursor, limit)); - } - EventRequest::NotifyUpdate { seq, tx } => { - let _ = tx.send(state.event_ops.notify_update(seq)); } } } @@ -5804,7 +5795,7 @@ .event_ops .insert_account_event(&did, AccountStatus::Deleted) }); - let _ = tx.send(result.map(|seq| seq.as_i64())); + let _ = tx.send(result.map(|_| ())); } UserRequest::CreatePasswordAccount { input, tx } => { let result = user.create_password_account(&input).and_then(|result| { diff --git a/crates/tranquil-api/src/identity/plc/submit.rs b/crates/tranquil-api/src/identity/plc/submit.rs --- a/crates/tranquil-api/src/identity/plc/submit.rs +++ b/crates/tranquil-api/src/identity/plc/submit.rs @@ -129,20 +129,13 @@ .await .map_err(ApiError::from)?; - match state + if let Err(e) = state .repos .repo .insert_identity_event(did, Some(&user.handle)) .await { - Ok(seq) => { - if let Err(e) = state.repos.repo.notify_update(seq).await { - warn!("Failed to notify identity event: {:?}", e); - } - } - Err(e) => { - warn!("Failed to sequence identity event: {:?}", e); - } + warn!("Failed to sequence identity event: {:?}", e); } let _ = state .cache