feat(tranquil-store): whole test suite working

Lewis: May this revision serve well! <lu5a@proton.me>
This commit is contained in:
Lewis
2026-04-10 13:57:44 +03:00
parent 1720de8650
commit dcd1593b1c
113 changed files with 5655 additions and 2928 deletions
@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT t.token FROM plc_operation_tokens t JOIN users u ON t.user_id = u.id WHERE u.did = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "token",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false
]
},
"hash": "05fd99170e31e68fa5028c862417cdf535cd70e09fde0a8a28249df0070eb2fc"
}
@@ -1,15 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE users SET deactivated_at = $1 WHERE did = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Timestamptz",
"Text"
]
},
"nullable": []
},
"hash": "0710b57fb9aa933525f617b15e6e2e5feaa9c59c38ec9175568abdacda167107"
}
@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT body FROM comms_queue WHERE user_id = (SELECT id FROM users WHERE did = $1) AND comms_type = 'email_update' ORDER BY created_at DESC LIMIT 1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "body",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false
]
},
"hash": "0ec60bb854a4991d0d7249a68f7445b65c8cc8c723baca221d85f5e4f2478b99"
}
@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT token FROM sso_pending_registration WHERE token = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "token",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false
]
},
"hash": "0fae1be7a75bdc58c69a9af97cad4aec23c32a9378764b8d6d7eb2cc89c562b1"
}
@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n DELETE FROM sso_pending_registration\n WHERE token = $1 AND expires_at > NOW()\n RETURNING token\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "token",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false
]
},
"hash": "1c84643fd6bc57c76517849a64d2d877df337e823d4c2c2b077f695bbfc9e9ac"
}
@@ -1,14 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE users SET is_admin = TRUE WHERE did = $1",
"query": "UPDATE users SET is_admin = $1 WHERE did = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Bool",
"Text"
]
},
"nullable": []
},
"hash": "f29da3bdfbbc547b339b4cdb059fac26435b0feec65cf1c56f851d1c4d6b1814"
"hash": "1e63d287c619a14e5c07d80e8e54193d2964c8b5e6a855256cb80c4d0cd2c6ea"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT token, expires_at FROM account_deletion_requests WHERE did = $1",
"query": "SELECT t.token, t.expires_at\n FROM plc_operation_tokens t\n JOIN users u ON t.user_id = u.id\n WHERE u.did = $1",
"describe": {
"columns": [
{
@@ -24,5 +24,5 @@
false
]
},
"hash": "47fe4a54857344d8f789f37092a294cd58f64b4fb431b54b5deda13d64525e88"
"hash": "237c2d912e89b7e0e5baa83503a22f158ea1614b5157f6c9e2aba6017fef6b26"
}
@@ -1,28 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n DELETE FROM sso_pending_registration\n WHERE token = $1 AND expires_at > NOW()\n RETURNING token, request_uri\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "token",
"type_info": "Text"
},
{
"ordinal": 1,
"name": "request_uri",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false,
false
]
},
"hash": "24b823043ab60f36c29029137fef30dfe33922bb06067f2fdbfc1fbb4b0a2a81"
}
@@ -1,38 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO external_identities (did, provider, provider_user_id)\n VALUES ($1, $2, $3)\n RETURNING id\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Uuid"
}
],
"parameters": {
"Left": [
"Text",
{
"Custom": {
"name": "sso_provider_type",
"kind": {
"Enum": [
"github",
"discord",
"google",
"gitlab",
"oidc",
"apple"
]
}
}
},
"Text"
]
},
"nullable": [
false
]
},
"hash": "2841093a67480e75e1e9e4046bf3eb74afae2d04f5ea0ec17a4d433983e6d71c"
}
@@ -1,32 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO sso_pending_registration (token, request_uri, provider, provider_user_id, provider_email_verified)\n VALUES ($1, $2, $3, $4, $5)\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
{
"Custom": {
"name": "sso_provider_type",
"kind": {
"Enum": [
"github",
"discord",
"google",
"gitlab",
"oidc",
"apple"
]
}
}
},
"Text",
"Bool"
]
},
"nullable": []
},
"hash": "376b72306b50f747bc9161985ff4f50c35c53025a55ccf5e9933dc3795d29313"
}
@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT id FROM external_identities WHERE id = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Uuid"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
false
]
},
"hash": "3933ea5b147ab6294936de147b98e116cfae848ecd76ea5d367585eb5117f2ad"
}
@@ -1,16 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO users (did, handle, email, password_hash) VALUES ($1, $2, $3, 'hash')",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
"Text"
]
},
"nullable": []
},
"hash": "3bed8d4843545f4a9676207513806603c50eb2af92957994abaf1c89c0294c12"
}
@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT id FROM users WHERE email = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Uuid"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false
]
},
"hash": "4560c237741ce9d4166aecd669770b3360a3ac71e649b293efb88d92c3254068"
}
@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT email_verified FROM users WHERE did = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "email_verified",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false
]
},
"hash": "49cbc923cc4a0dcf7dea4ead5ab9580ff03b717586c4ca2d5343709e2dac86b6"
}
@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT token FROM sso_pending_registration\n WHERE token = $1 AND expires_at > NOW()\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "token",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false
]
},
"hash": "4fef326fa2d03d04869af3fec702c901d1ecf392545a3a032438b2c1859d46cc"
}
@@ -1,15 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO oauth_authorization_request (id, client_id, parameters, expires_at)\n VALUES ($1, 'https://test.example.com', $2, NOW() + INTERVAL '1 hour')\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Jsonb"
]
},
"nullable": []
},
"hash": "575c1e5529874f8f523e6fe22ccf4ee3296806581b1765dfb91a84ffab347f15"
}
@@ -1,33 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO sso_pending_registration (token, request_uri, provider, provider_user_id, provider_username, provider_email)\n VALUES ($1, $2, $3, $4, $5, $6)\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
{
"Custom": {
"name": "sso_provider_type",
"kind": {
"Enum": [
"github",
"discord",
"google",
"gitlab",
"oidc",
"apple"
]
}
}
},
"Text",
"Text",
"Text"
]
},
"nullable": []
},
"hash": "596c3400a60c77c7645fd46fcea61fa7898b6832e58c0f647f382b23b81d350e"
}
@@ -1,81 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT id, did, provider as \"provider: SsoProviderType\", provider_user_id, provider_username, provider_email\n FROM external_identities\n WHERE provider = $1 AND provider_user_id = $2\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Uuid"
},
{
"ordinal": 1,
"name": "did",
"type_info": "Text"
},
{
"ordinal": 2,
"name": "provider: SsoProviderType",
"type_info": {
"Custom": {
"name": "sso_provider_type",
"kind": {
"Enum": [
"github",
"discord",
"google",
"gitlab",
"oidc",
"apple"
]
}
}
}
},
{
"ordinal": 3,
"name": "provider_user_id",
"type_info": "Text"
},
{
"ordinal": 4,
"name": "provider_username",
"type_info": "Text"
},
{
"ordinal": 5,
"name": "provider_email",
"type_info": "Text"
}
],
"parameters": {
"Left": [
{
"Custom": {
"name": "sso_provider_type",
"kind": {
"Enum": [
"github",
"discord",
"google",
"gitlab",
"oidc",
"apple"
]
}
}
},
"Text"
]
},
"nullable": [
false,
false,
false,
false,
true,
true
]
},
"hash": "59e63c5cf92985714e9586d1ce012efef733d4afaa4ea09974daf8303805e5d2"
}
@@ -1,28 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT k.key_bytes, k.encryption_version\n FROM user_keys k\n JOIN users u ON k.user_id = u.id\n WHERE u.did = $1\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "key_bytes",
"type_info": "Bytea"
},
{
"ordinal": 1,
"name": "encryption_version",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false,
true
]
},
"hash": "5a016f289caf75177731711e56e92881ba343c73a9a6e513e205c801c5943ec0"
}
@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n DELETE FROM sso_auth_state\n WHERE state = $1 AND expires_at > NOW()\n RETURNING state\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "state",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false
]
},
"hash": "5af4a386c1632903ad7102551a5bd148bcf541baab6a84c8649666a695f9c4d1"
}
@@ -1,11 +1,11 @@
{
"db_name": "PostgreSQL",
"query": "SELECT token FROM account_deletion_requests WHERE did = $1",
"query": "SELECT code FROM oauth_2fa_challenge WHERE request_uri = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "token",
"name": "code",
"type_info": "Text"
}
],
@@ -18,5 +18,5 @@
false
]
},
"hash": "a802d7d860f263eace39ce82bb27b633cec7287c1cc177f0e1d47ec6571564d5"
"hash": "5cee16f49a727d66b5231a8d07d7f4bcb6a1136fbf3e3d249fd33600772ac80f"
}
@@ -1,43 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT provider_user_id, provider_email_verified\n FROM external_identities\n WHERE did = $1 AND provider = $2\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "provider_user_id",
"type_info": "Text"
},
{
"ordinal": 1,
"name": "provider_email_verified",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Text",
{
"Custom": {
"name": "sso_provider_type",
"kind": {
"Enum": [
"github",
"discord",
"google",
"gitlab",
"oidc",
"apple"
]
}
}
}
]
},
"nullable": [
false,
false
]
},
"hash": "5e4c0dd92ac3c4b5e2eae5d129f2649cf3a8f068105f44a8dca9625427affc06"
}
@@ -1,33 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO sso_auth_state (state, request_uri, provider, action, nonce, code_verifier)\n VALUES ($1, $2, $3, $4, $5, $6)\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
{
"Custom": {
"name": "sso_provider_type",
"kind": {
"Enum": [
"github",
"discord",
"google",
"gitlab",
"oidc",
"apple"
]
}
}
},
"Text",
"Text",
"Text"
]
},
"nullable": []
},
"hash": "5e9c6ec72c2c0ea1c8dff551d01baddd1dd953c828a5656db2ee39dea996f890"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT t.token, t.expires_at FROM plc_operation_tokens t JOIN users u ON t.user_id = u.id WHERE u.did = $1",
"query": "SELECT token, did, expires_at FROM account_deletion_requests WHERE did = $1",
"describe": {
"columns": [
{
@@ -10,6 +10,11 @@
},
{
"ordinal": 1,
"name": "did",
"type_info": "Text"
},
{
"ordinal": 2,
"name": "expires_at",
"type_info": "Timestamptz"
}
@@ -20,9 +25,10 @@
]
},
"nullable": [
false,
false,
false
]
},
"hash": "82717b6f61cd79347e1ca7e92c4413743ba168d1e0d8b85566711e54d4048f81"
"hash": "61f489b4fc42f5b0aaea287cde4415da6f5e96b3a0f36216bdc6dea924b09abd"
}
@@ -1,28 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT did, email_verified FROM users WHERE did = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "did",
"type_info": "Text"
},
{
"ordinal": 1,
"name": "email_verified",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false,
false
]
},
"hash": "63f6f2a89650794fe90e10ce7fc785a6b9f7d37c12b31a6ff13f7c5214eef19e"
}
@@ -1,66 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT state, request_uri, provider as \"provider: SsoProviderType\", action, nonce, code_verifier\n FROM sso_auth_state\n WHERE state = $1\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "state",
"type_info": "Text"
},
{
"ordinal": 1,
"name": "request_uri",
"type_info": "Text"
},
{
"ordinal": 2,
"name": "provider: SsoProviderType",
"type_info": {
"Custom": {
"name": "sso_provider_type",
"kind": {
"Enum": [
"github",
"discord",
"google",
"gitlab",
"oidc",
"apple"
]
}
}
}
},
{
"ordinal": 3,
"name": "action",
"type_info": "Text"
},
{
"ordinal": 4,
"name": "nonce",
"type_info": "Text"
},
{
"ordinal": 5,
"name": "code_verifier",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false,
false,
false,
false,
true,
true
]
},
"hash": "6c7ace2a64848adc757af6c93b9162e1d95788b372370a7ad0d7540338bb73ee"
}
@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT state FROM sso_auth_state WHERE state = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "state",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false
]
},
"hash": "6fbcff0206599484bfb6cef165b6f729d27e7a342f7718ee4ac07f0ca94412ba"
}
@@ -1,33 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO sso_pending_registration (token, request_uri, provider, provider_user_id, provider_username, provider_email_verified)\n VALUES ($1, $2, $3, $4, $5, $6)\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
{
"Custom": {
"name": "sso_provider_type",
"kind": {
"Enum": [
"github",
"discord",
"google",
"gitlab",
"oidc",
"apple"
]
}
}
},
"Text",
"Text",
"Bool"
]
},
"nullable": []
},
"hash": "712459c27fc037f45389e2766cf1057e86e93ef756a784ed12beb453b03c5da1"
}
@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT subject FROM comms_queue WHERE user_id = $1 AND comms_type = 'admin_email' AND body = 'Email without subject' LIMIT 1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "subject",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
true
]
},
"hash": "785a864944c5939331704c71b0cd3ed26ffdd64f3fd0f26ecc28b6a0557bbe8f"
}
@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT t.token\n FROM plc_operation_tokens t\n JOIN users u ON t.user_id = u.id\n WHERE u.did = $1\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "token",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false
]
},
"hash": "7caa8f9083b15ec1209dda35c4c6f6fba9fe338e4a6a10636b5389d426df1631"
}
@@ -1,28 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT provider_username, last_login_at FROM external_identities WHERE id = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "provider_username",
"type_info": "Text"
},
{
"ordinal": 1,
"name": "last_login_at",
"type_info": "Timestamptz"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
true,
true
]
},
"hash": "7d24e744a4e63570b1410e50b45b745ce8915ab3715b3eff7efc2d84f27735d0"
}
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO user_totp (did, secret_encrypted, encryption_version, verified, created_at)\n VALUES ($1, $2, 1, TRUE, NOW())\n ON CONFLICT (did) DO UPDATE SET secret_encrypted = $2, verified = TRUE",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Bytea"
]
},
"nullable": []
},
"hash": "7d3a9f0545943bc6a3a14fcd596aac5cc731c8177d74e504606d7e92c7d0c73f"
}
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE users SET two_factor_enabled = $1 WHERE did = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Bool",
"Text"
]
},
"nullable": []
},
"hash": "84a1db51a98402323cb86bc19cd2b737f908222ea3426b8bf47d735aff5b6c75"
}
@@ -1,34 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO sso_pending_registration (token, request_uri, provider, provider_user_id, provider_username, provider_email, provider_email_verified)\n VALUES ($1, $2, $3, $4, $5, $6, $7)\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
{
"Custom": {
"name": "sso_provider_type",
"kind": {
"Enum": [
"github",
"discord",
"google",
"gitlab",
"oidc",
"apple"
]
}
}
},
"Text",
"Text",
"Text",
"Bool"
]
},
"nullable": []
},
"hash": "85ffc37a77af832d7795f5f37efe304fced4bf56b4f2287fe9aeb3fc97e1b191"
}
@@ -0,0 +1,36 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM comms_queue WHERE user_id = $1 AND comms_type = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid",
{
"Custom": {
"name": "comms_type",
"kind": {
"Enum": [
"welcome",
"email_verification",
"password_reset",
"email_update",
"account_deletion",
"admin_email",
"plc_operation",
"two_factor_code",
"channel_verification",
"passkey_recovery",
"legacy_login_alert",
"migration_verification",
"channel_verified"
]
}
}
}
]
},
"nullable": []
},
"hash": "89b0292d8d022fad8f9cda07b9a7870ca6a7ebe904b2d580956b0816b50bcdb7"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT COUNT(*) as \"count!\" FROM plc_operation_tokens t JOIN users u ON t.user_id = u.id WHERE u.did = $1",
"query": "SELECT COUNT(*) as \"count!\"\n FROM plc_operation_tokens t\n JOIN users u ON t.user_id = u.id\n WHERE u.did = $1",
"describe": {
"columns": [
{
@@ -18,5 +18,5 @@
null
]
},
"hash": "cda68f9b6c60295a196fc853b70ec5fd51a8ffaa2bac5942c115c99d1cbcafa3"
"hash": "990bf50e60fc5566639c2c12cd968d154d7b0c6863ad69141653135f98fbc998"
}
@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT body FROM comms_queue WHERE user_id = (SELECT id FROM users WHERE did = $1) AND comms_type = 'email_verification' ORDER BY created_at DESC LIMIT 1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "body",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false
]
},
"hash": "9ad422bf3c43e3cfd86fc88c73594246ead214ca794760d3fe77bb5cf4f27be5"
}
@@ -1,28 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT did, public_key_did_key FROM reserved_signing_keys WHERE public_key_did_key = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "did",
"type_info": "Text"
},
{
"ordinal": 1,
"name": "public_key_did_key",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
true,
false
]
},
"hash": "9b035b051769e6b9d45910a8bb42ac0f84c73de8c244ba4560f004ee3f4b7002"
}
@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT id FROM external_identities WHERE did = $1\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Uuid"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false
]
},
"hash": "9dba64081d4f95b5490c9a9bf30a7175db3429f39df4f25e212f38f33882fc65"
}
@@ -0,0 +1,180 @@
{
"db_name": "PostgreSQL",
"query": "SELECT\n id, user_id,\n channel as \"channel: CommsChannel\",\n comms_type as \"comms_type: CommsType\",\n status as \"status: CommsStatus\",\n recipient, subject, body, metadata,\n attempts, max_attempts, last_error,\n created_at, updated_at, scheduled_for, processed_at\n FROM comms_queue\n WHERE user_id = $1 AND comms_type = $2\n ORDER BY created_at DESC\n LIMIT $3",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Uuid"
},
{
"ordinal": 1,
"name": "user_id",
"type_info": "Uuid"
},
{
"ordinal": 2,
"name": "channel: CommsChannel",
"type_info": {
"Custom": {
"name": "comms_channel",
"kind": {
"Enum": [
"email",
"discord",
"telegram",
"signal"
]
}
}
}
},
{
"ordinal": 3,
"name": "comms_type: CommsType",
"type_info": {
"Custom": {
"name": "comms_type",
"kind": {
"Enum": [
"welcome",
"email_verification",
"password_reset",
"email_update",
"account_deletion",
"admin_email",
"plc_operation",
"two_factor_code",
"channel_verification",
"passkey_recovery",
"legacy_login_alert",
"migration_verification",
"channel_verified"
]
}
}
}
},
{
"ordinal": 4,
"name": "status: CommsStatus",
"type_info": {
"Custom": {
"name": "comms_status",
"kind": {
"Enum": [
"pending",
"processing",
"sent",
"failed"
]
}
}
}
},
{
"ordinal": 5,
"name": "recipient",
"type_info": "Text"
},
{
"ordinal": 6,
"name": "subject",
"type_info": "Text"
},
{
"ordinal": 7,
"name": "body",
"type_info": "Text"
},
{
"ordinal": 8,
"name": "metadata",
"type_info": "Jsonb"
},
{
"ordinal": 9,
"name": "attempts",
"type_info": "Int4"
},
{
"ordinal": 10,
"name": "max_attempts",
"type_info": "Int4"
},
{
"ordinal": 11,
"name": "last_error",
"type_info": "Text"
},
{
"ordinal": 12,
"name": "created_at",
"type_info": "Timestamptz"
},
{
"ordinal": 13,
"name": "updated_at",
"type_info": "Timestamptz"
},
{
"ordinal": 14,
"name": "scheduled_for",
"type_info": "Timestamptz"
},
{
"ordinal": 15,
"name": "processed_at",
"type_info": "Timestamptz"
}
],
"parameters": {
"Left": [
"Uuid",
{
"Custom": {
"name": "comms_type",
"kind": {
"Enum": [
"welcome",
"email_verification",
"password_reset",
"email_update",
"account_deletion",
"admin_email",
"plc_operation",
"two_factor_code",
"channel_verification",
"passkey_recovery",
"legacy_login_alert",
"migration_verification",
"channel_verified"
]
}
}
},
"Int8"
]
},
"nullable": [
false,
false,
false,
false,
false,
false,
true,
false,
true,
false,
false,
true,
false,
false,
false,
true
]
},
"hash": "9f3f2b36f11e9446915d3ca29ef81e4ada0c6a6d72764116dac4f99a4e09785e"
}
@@ -1,66 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT token, request_uri, provider as \"provider: SsoProviderType\", provider_user_id,\n provider_username, provider_email\n FROM sso_pending_registration\n WHERE token = $1 AND expires_at > NOW()\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "token",
"type_info": "Text"
},
{
"ordinal": 1,
"name": "request_uri",
"type_info": "Text"
},
{
"ordinal": 2,
"name": "provider: SsoProviderType",
"type_info": {
"Custom": {
"name": "sso_provider_type",
"kind": {
"Enum": [
"github",
"discord",
"google",
"gitlab",
"oidc",
"apple"
]
}
}
}
},
{
"ordinal": 3,
"name": "provider_user_id",
"type_info": "Text"
},
{
"ordinal": 4,
"name": "provider_username",
"type_info": "Text"
},
{
"ordinal": 5,
"name": "provider_email",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false,
false,
false,
false,
true,
true
]
},
"hash": "9fd56986c1c843d386d1e5884acef8573eb55a3e9f5cb0122fcf8b93d6d667a5"
}
@@ -1,34 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT private_key_bytes, expires_at, used_at FROM reserved_signing_keys WHERE public_key_did_key = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "private_key_bytes",
"type_info": "Bytea"
},
{
"ordinal": 1,
"name": "expires_at",
"type_info": "Timestamptz"
},
{
"ordinal": 2,
"name": "used_at",
"type_info": "Timestamptz"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false,
false,
true
]
},
"hash": "a23a390659616779d7dbceaa3b5d5171e70fa25e3b8393e142cebcbff752f0f5"
}
@@ -1,15 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE external_identities\n SET provider_username = $2, last_login_at = NOW()\n WHERE id = $1\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": []
},
"hash": "a3d549a32e76c24e265c73a98dd739067623f275de0740bd576ee288f4444496"
}
@@ -1,40 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT preferred_comms_channel as \"preferred_comms_channel: String\", discord_username FROM users WHERE did = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "preferred_comms_channel: String",
"type_info": {
"Custom": {
"name": "comms_channel",
"kind": {
"Enum": [
"email",
"discord",
"telegram",
"signal"
]
}
}
}
},
{
"ordinal": 1,
"name": "discord_username",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false,
true
]
},
"hash": "a844774d8dd3c50c5faf3de5d43f534b80234759c8437434e467ca33ea10fd1f"
}
@@ -1,28 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n DELETE FROM sso_auth_state\n WHERE state = $1 AND expires_at > NOW()\n RETURNING state, request_uri\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "state",
"type_info": "Text"
},
{
"ordinal": 1,
"name": "request_uri",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false,
false
]
},
"hash": "aee3e8e1d8924d41bec7d866e274f8bb2ddef833eb03326103c2d0a17ee56154"
}
@@ -1,22 +1,18 @@
{
"db_name": "PostgreSQL",
"query": "SELECT subject, body, comms_type as \"comms_type: String\" FROM comms_queue WHERE user_id = $1 AND comms_type = 'admin_email' ORDER BY created_at DESC LIMIT 1",
"query": "SELECT COUNT(*) as \"count!\" FROM comms_queue WHERE user_id = $1 AND comms_type = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "subject",
"type_info": "Text"
},
{
"ordinal": 1,
"name": "body",
"type_info": "Text"
},
{
"ordinal": 2,
"name": "comms_type: String",
"type_info": {
"name": "count!",
"type_info": "Int8"
}
],
"parameters": {
"Left": [
"Uuid",
{
"Custom": {
"name": "comms_type",
"kind": {
@@ -38,18 +34,11 @@
}
}
}
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
true,
false,
false
null
]
},
"hash": "4445cc86cdf04894b340e67661b79a3c411917144a011f50849b737130b24dbe"
"hash": "b364a2b202bab17c0cdc5f70d23b13841b4d9063d94cd0b09268c3dc41824fd2"
}
@@ -1,31 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO sso_pending_registration (token, request_uri, provider, provider_user_id, expires_at)\n VALUES ($1, $2, $3, $4, NOW() - INTERVAL '1 hour')\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
{
"Custom": {
"name": "sso_provider_type",
"kind": {
"Enum": [
"github",
"discord",
"google",
"gitlab",
"oidc",
"apple"
]
}
}
},
"Text"
]
},
"nullable": []
},
"hash": "ba9684872fad5201b8504c2606c29364a2df9631fe98817e7bfacd3f3f51f6cb"
}
@@ -1,12 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM sso_auth_state WHERE expires_at < NOW()",
"describe": {
"columns": [],
"parameters": {
"Left": []
},
"nullable": []
},
"hash": "bb4460f75d30f48b79d71b97f2c7d54190260deba2d2ade177dbdaa507ab275b"
}
@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT password_reset_code FROM users WHERE email = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "password_reset_code",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
true
]
},
"hash": "cd3b8098ad4c1056c1d23acd8a6b29f7abfe18ee6f559bd94ab16274b1cfdfee"
}
@@ -1,31 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO sso_auth_state (state, request_uri, provider, action, expires_at)\n VALUES ($1, $2, $3, $4, NOW() - INTERVAL '1 hour')\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
{
"Custom": {
"name": "sso_provider_type",
"kind": {
"Enum": [
"github",
"discord",
"google",
"gitlab",
"oidc",
"apple"
]
}
}
},
"Text"
]
},
"nullable": []
},
"hash": "d0d4fb4b44cda3442b20037b4d5efaa032e1d004c775e2b6077c5050d7d62041"
}
@@ -1,40 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO external_identities (did, provider, provider_user_id, provider_username, provider_email)\n VALUES ($1, $2, $3, $4, $5)\n RETURNING id\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Uuid"
}
],
"parameters": {
"Left": [
"Text",
{
"Custom": {
"name": "sso_provider_type",
"kind": {
"Enum": [
"github",
"discord",
"google",
"gitlab",
"oidc",
"apple"
]
}
}
},
"Text",
"Text",
"Text"
]
},
"nullable": [
false
]
},
"hash": "dd7d80d4d118a5fc95b574e2ca9ffaccf974e52fb6ac368f716409c55f9d3ab0"
}
@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT COUNT(*) FROM comms_queue WHERE user_id = $1 AND comms_type = 'password_reset'",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "count",
"type_info": "Int8"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
null
]
},
"hash": "e20cbe2a939d790aaea718b084a80d8ede655ba1cc0fd4346d7e91d6de7d6cf3"
}
@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT used_at FROM reserved_signing_keys WHERE public_key_did_key = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "used_at",
"type_info": "Timestamptz"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
true
]
},
"hash": "e64cd36284d10ab7f3d9f6959975a1a627809f444b0faff7e611d985f31b90e9"
}
@@ -1,30 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO external_identities (did, provider, provider_user_id)\n VALUES ($1, $2, $3)\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
{
"Custom": {
"name": "sso_provider_type",
"kind": {
"Enum": [
"github",
"discord",
"google",
"gitlab",
"oidc",
"apple"
]
}
}
},
"Text"
]
},
"nullable": []
},
"hash": "eb54d2ce02cab7c2e7f9926bd469b19e5f0513f47173b2738fc01a57082d7abb"
}
@@ -1,15 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM external_identities WHERE id = $1 AND did = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": []
},
"hash": "ec22a8cc89e480c403a239eac44288e144d83364129491de6156760616666d3d"
}
@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT email FROM users WHERE did = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "email",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
true
]
},
"hash": "f26c13023b47b908ec96da2e6b8bf8b34ca6a2246c20fc96f76f0e95530762a7"
}
@@ -0,0 +1,52 @@
{
"db_name": "PostgreSQL",
"query": "SELECT id, did, public_key_did_key, private_key_bytes, expires_at, used_at\n FROM reserved_signing_keys WHERE public_key_did_key = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Uuid"
},
{
"ordinal": 1,
"name": "did",
"type_info": "Text"
},
{
"ordinal": 2,
"name": "public_key_did_key",
"type_info": "Text"
},
{
"ordinal": 3,
"name": "private_key_bytes",
"type_info": "Bytea"
},
{
"ordinal": 4,
"name": "expires_at",
"type_info": "Timestamptz"
},
{
"ordinal": 5,
"name": "used_at",
"type_info": "Timestamptz"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false,
true,
false,
false,
false,
true
]
},
"hash": "f3b07f153284b6dd1f22c098af8628d85dcdf20dd6273443ff98d02b6f5ecbf1"
}
Generated
+2
View File
@@ -7715,6 +7715,7 @@ dependencies = [
"tower-http",
"tower-layer",
"tracing",
"tracing-subscriber",
"tranquil-api",
"tranquil-auth",
"tranquil-cache",
@@ -7821,6 +7822,7 @@ version = "0.4.7"
dependencies = [
"async-trait",
"chrono",
"fjall",
"futures",
"presage",
"rand 0.9.2",
+2 -2
View File
@@ -26,7 +26,7 @@ members = [
]
[workspace.package]
version = "0.4.7"
version = "0.5.0"
edition = "2024"
license = "AGPL-3.0-or-later"
@@ -51,7 +51,7 @@ tranquil-server = { path = "crates/tranquil-server" }
tranquil-sync = { path = "crates/tranquil-sync" }
tranquil-oauth-server = { path = "crates/tranquil-oauth-server" }
tranquil-api = { path = "crates/tranquil-api" }
tranquil-signal = { path = "crates/tranquil-signal" }
tranquil-signal = { path = "crates/tranquil-signal", features = ["fjall-store"] }
tranquil-store = { path = "crates/tranquil-store" }
presage = { git = "https://github.com/whisperfish/presage", rev = "fe3ed54c4844ae51c3a9fa49cf80a7816a31a425", default-features = false }
+1
View File
@@ -29,6 +29,7 @@ COPY crates/tranquil-pds ./crates/tranquil-pds
COPY crates/tranquil-sync ./crates/tranquil-sync
COPY crates/tranquil-api ./crates/tranquil-api
COPY crates/tranquil-oauth-server ./crates/tranquil-oauth-server
COPY crates/tranquil-store ./crates/tranquil-store
COPY crates/tranquil-signal ./crates/tranquil-signal
COPY crates/tranquil-server ./crates/tranquil-server
COPY migrations ./crates/tranquil-pds/migrations
@@ -51,16 +51,14 @@ pub async fn search_accounts(
Query(params): Query<SearchAccountsParams>,
) -> Result<Json<SearchAccountsOutput>, ApiError> {
let limit = params.limit.clamp(1, 100);
let email_filter = params.email.as_deref().map(|e| format!("%{}%", e));
let handle_filter = params.handle.as_deref().map(|h| format!("%{}%", h));
let cursor_did: Option<Did> = params.cursor.as_ref().and_then(|c| c.parse().ok());
let rows = state
.repos
.user
.search_accounts(
cursor_did.as_ref(),
email_filter.as_deref(),
handle_filter.as_deref(),
params.email.as_deref(),
params.handle.as_deref(),
limit + 1,
)
.await
+21 -13
View File
@@ -5,7 +5,6 @@ use serde::Serialize;
use tranquil_pds::api::error::ApiError;
use tranquil_pds::auth::{Admin, Auth};
use tranquil_pds::state::AppState;
use tranquil_signal::PgSignalStore;
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
@@ -53,15 +52,20 @@ pub async fn link_signal_device(
let device_name = tranquil_signal::DeviceName::new("tranquil-pds".to_string())
.map_err(|e| ApiError::InternalError(Some(format!("invalid device name: {e}"))))?;
let link_result = tranquil_signal::SignalClient::link_device(
&state.repos.pool,
device_name,
state.shutdown.clone(),
link_cancel,
slot.linking_flag(),
)
.await
.map_err(|e| ApiError::InternalError(Some(format!("Signal linking failed: {e}"))))?;
let signal_store = state
.signal_store_provider
.as_ref()
.ok_or_else(|| ApiError::InternalError(Some("Signal store not configured".into())))?;
let link_result = signal_store
.link_signal_device(
device_name,
state.shutdown.clone(),
link_cancel,
slot.linking_flag(),
)
.await
.map_err(|e| ApiError::InternalError(Some(format!("Signal linking failed: {e}"))))?;
let qr_base64 = url_to_qr_png_base64(link_result.url.as_str())
.map_err(|e| ApiError::InternalError(Some(format!("QR generation failed: {e}"))))?;
@@ -108,9 +112,13 @@ pub async fn unlink_signal_device(
.as_ref()
.ok_or_else(|| ApiError::InvalidRequest("Signal is not enabled".into()))?;
let store = PgSignalStore::new(state.repos.pool.clone());
store
.clear_all()
let signal_store = state
.signal_store_provider
.as_ref()
.ok_or_else(|| ApiError::InternalError(Some("Signal store not configured".into())))?;
signal_store
.clear_signal_data()
.await
.map_err(|e| ApiError::InternalError(Some(format!("Failed to clear signal data: {e}"))))?;
-17
View File
@@ -18,7 +18,6 @@ use tranquil_pds::delegation::{
use tranquil_pds::rate_limit::{AccountCreationLimit, RateLimited};
use tranquil_pds::state::AppState;
use tranquil_pds::types::{Did, Handle};
use tranquil_types::CidLink;
pub async fn list_controllers(
State(state): State<AppState>,
@@ -420,22 +419,6 @@ pub async fn create_delegated_account(
}
};
state
.repos
.repo
.create_repo(
user_id,
&did,
&handle,
&CidLink::from(&repo.commit_cid),
&repo.repo_rev,
)
.await
.map_err(|e| {
error!("failed to register repo in backend: {e:?}");
ApiError::InternalError(None)
})?;
if let Some(validated) = validated_invite_code
&& let Err(e) = state
.repos
@@ -15,8 +15,6 @@ use tranquil_pds::rate_limit::{AccountCreationLimit, RateLimited};
use tranquil_pds::state::AppState;
use tranquil_pds::types::{Did, Handle, PlainPassword};
use tranquil_pds::validation::validate_password;
use tranquil_types::CidLink;
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CreateAccountInput {
@@ -548,21 +546,6 @@ pub async fn create_account(
}
};
let user_id = create_result.user_id;
if let Err(e) = state
.repos
.repo
.create_repo(
user_id,
&did_for_commit,
&handle_typed,
&CidLink::from(&repo.commit_cid),
&repo.repo_rev,
)
.await
{
error!("failed to register repo in backend: {e:?}");
return ApiError::InternalError(None).into_response();
}
if !is_migration && !is_did_web_byod {
super::provision::sequence_new_account(
&state,
@@ -15,7 +15,6 @@ use tranquil_pds::rate_limit::{AccountCreationLimit, PasswordResetLimit, RateLim
use tranquil_pds::state::AppState;
use tranquil_pds::types::{Did, Handle, PlainPassword};
use tranquil_pds::validation::validate_password;
use tranquil_types::CidLink;
fn generate_setup_token() -> String {
let mut rng = rand::thread_rng();
@@ -367,22 +366,6 @@ pub async fn create_passkey_account(
};
let user_id = create_result.user_id;
state
.repos
.repo
.create_repo(
user_id,
&did_typed,
&handle_typed,
&CidLink::from(&repo.commit_cid),
&repo.repo_rev,
)
.await
.map_err(|e| {
error!("failed to register repo in backend: {e:?}");
ApiError::InternalError(None)
})?;
if !is_byod_did_web {
crate::identity::provision::sequence_new_account(
&state,
+64
View File
@@ -185,12 +185,41 @@ pub struct ReservedSigningKey {
pub private_key_bytes: Vec<u8>,
}
#[derive(Debug, Clone)]
pub struct ReservedSigningKeyFull {
pub id: Uuid,
pub did: Option<Did>,
pub public_key_did_key: String,
pub private_key_bytes: Vec<u8>,
pub expires_at: DateTime<Utc>,
pub used_at: Option<DateTime<Utc>>,
}
#[derive(Debug, Clone)]
pub struct DeletionRequest {
pub did: Did,
pub expires_at: DateTime<Utc>,
}
#[derive(Debug, Clone)]
pub struct DeletionRequestWithToken {
pub token: String,
pub did: Did,
pub expires_at: DateTime<Utc>,
}
#[derive(Debug, Clone)]
pub struct PlcTokenInfo {
pub token: String,
pub expires_at: DateTime<Utc>,
}
#[derive(Debug, Clone)]
pub struct PasswordResetInfo {
pub code: Option<String>,
pub expires_at: Option<DateTime<Utc>>,
}
#[async_trait]
pub trait InfraRepository: Send + Sync {
#[allow(clippy::too_many_arguments)]
@@ -406,6 +435,41 @@ pub trait InfraRepository: Send + Sync {
&self,
user_ids: &[Uuid],
) -> Result<Vec<(Uuid, String)>, DbError>;
async fn get_deletion_request_by_did(
&self,
did: &Did,
) -> Result<Option<DeletionRequestWithToken>, DbError>;
async fn get_latest_comms_for_user(
&self,
user_id: Uuid,
comms_type: CommsType,
limit: i64,
) -> Result<Vec<QueuedComms>, DbError>;
async fn count_comms_by_type(
&self,
user_id: Uuid,
comms_type: CommsType,
) -> Result<i64, DbError>;
async fn delete_comms_by_type_for_user(
&self,
user_id: Uuid,
comms_type: CommsType,
) -> Result<u64, DbError>;
async fn expire_deletion_request(&self, token: &str) -> Result<(), DbError>;
async fn get_reserved_signing_key_full(
&self,
public_key_did_key: &str,
) -> Result<Option<ReservedSigningKeyFull>, DbError>;
async fn get_plc_tokens_by_did(&self, did: &Did) -> Result<Vec<PlcTokenInfo>, DbError>;
async fn count_plc_tokens_by_did(&self, did: &Did) -> Result<i64, DbError>;
}
#[derive(Debug, Clone)]
+4 -3
View File
@@ -22,9 +22,10 @@ pub use delegation::{
};
pub use error::DbError;
pub use infra::{
AdminAccountInfo, CommsChannel, CommsStatus, CommsType, DeletionRequest, InfraRepository,
InviteCodeInfo, InviteCodeRow, InviteCodeSortOrder, InviteCodeState, InviteCodeUse,
NotificationHistoryRow, QueuedComms, ReservedSigningKey,
AdminAccountInfo, CommsChannel, CommsStatus, CommsType, DeletionRequest,
DeletionRequestWithToken, InfraRepository, InviteCodeInfo, InviteCodeRow, InviteCodeSortOrder,
InviteCodeState, InviteCodeUse, NotificationHistoryRow, PasswordResetInfo, PlcTokenInfo,
QueuedComms, ReservedSigningKey, ReservedSigningKeyFull,
};
pub use invite_code::{InviteCodeError, ValidatedInviteCode};
pub use oauth::{
+5
View File
@@ -324,4 +324,9 @@ pub trait OAuthRepository: Send + Sync {
did: &Did,
except_token_id: &TokenId,
) -> Result<u64, DbError>;
async fn get_2fa_challenge_code(
&self,
request_uri: &RequestId,
) -> Result<Option<String>, DbError>;
}
+14
View File
@@ -223,6 +223,8 @@ pub trait UserRepository: Send + Sync {
async fn admin_update_password(&self, did: &Did, password_hash: &str) -> Result<u64, DbError>;
async fn set_admin_status(&self, did: &Did, is_admin: bool) -> Result<(), DbError>;
async fn get_notification_prefs(&self, did: &Did)
-> Result<Option<NotificationPrefs>, DbError>;
@@ -584,6 +586,18 @@ pub trait UserRepository: Send + Sync {
&self,
input: &RecoverPasskeyAccountInput,
) -> Result<RecoverPasskeyAccountResult, DbError>;
async fn get_password_reset_info(
&self,
email: &str,
) -> Result<Option<crate::PasswordResetInfo>, DbError>;
async fn enable_totp_verified(&self, did: &Did, encrypted_secret: &[u8])
-> Result<(), DbError>;
async fn set_two_factor_enabled(&self, did: &Did, enabled: bool) -> Result<(), DbError>;
async fn expire_password_reset_code(&self, email: &str) -> Result<(), DbError>;
}
#[derive(Debug, Clone)]
+158 -3
View File
@@ -3,9 +3,9 @@ use chrono::{DateTime, Utc};
use sqlx::PgPool;
use tranquil_db_traits::{
AdminAccountInfo, CommsChannel, CommsStatus, CommsType, DbError, DeletionRequest,
InfraRepository, InviteCodeError, InviteCodeInfo, InviteCodeRow, InviteCodeSortOrder,
InviteCodeState, InviteCodeUse, NotificationHistoryRow, QueuedComms, ReservedSigningKey,
ValidatedInviteCode,
DeletionRequestWithToken, InfraRepository, InviteCodeError, InviteCodeInfo, InviteCodeRow,
InviteCodeSortOrder, InviteCodeState, InviteCodeUse, NotificationHistoryRow, PlcTokenInfo,
QueuedComms, ReservedSigningKey, ReservedSigningKeyFull, ValidatedInviteCode,
};
use tranquil_types::{CidLink, Did, Handle};
use uuid::Uuid;
@@ -1034,4 +1034,159 @@ impl InfraRepository for PostgresInfraRepository {
.map(|r| (r.used_by_user, r.code))
.collect())
}
async fn get_deletion_request_by_did(
&self,
did: &Did,
) -> Result<Option<DeletionRequestWithToken>, DbError> {
let row = sqlx::query!(
r#"SELECT token, did, expires_at FROM account_deletion_requests WHERE did = $1"#,
did.as_str()
)
.fetch_optional(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(row.map(|r| DeletionRequestWithToken {
token: r.token,
did: Did::new(r.did).expect("valid DID in database"),
expires_at: r.expires_at,
}))
}
async fn get_latest_comms_for_user(
&self,
user_id: Uuid,
comms_type: CommsType,
limit: i64,
) -> Result<Vec<QueuedComms>, DbError> {
let results = sqlx::query_as!(
QueuedComms,
r#"SELECT
id, user_id,
channel as "channel: CommsChannel",
comms_type as "comms_type: CommsType",
status as "status: CommsStatus",
recipient, subject, body, metadata,
attempts, max_attempts, last_error,
created_at, updated_at, scheduled_for, processed_at
FROM comms_queue
WHERE user_id = $1 AND comms_type = $2
ORDER BY created_at DESC
LIMIT $3"#,
user_id,
comms_type as CommsType,
limit
)
.fetch_all(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(results)
}
async fn count_comms_by_type(
&self,
user_id: Uuid,
comms_type: CommsType,
) -> Result<i64, DbError> {
let count = sqlx::query_scalar!(
r#"SELECT COUNT(*) as "count!" FROM comms_queue WHERE user_id = $1 AND comms_type = $2"#,
user_id,
comms_type as CommsType
)
.fetch_one(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(count)
}
async fn delete_comms_by_type_for_user(
&self,
user_id: Uuid,
comms_type: CommsType,
) -> Result<u64, DbError> {
let result = sqlx::query!(
"DELETE FROM comms_queue WHERE user_id = $1 AND comms_type = $2",
user_id,
comms_type as CommsType
)
.execute(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(result.rows_affected())
}
async fn expire_deletion_request(&self, token: &str) -> Result<(), DbError> {
sqlx::query!(
"UPDATE account_deletion_requests SET expires_at = NOW() - INTERVAL '1 hour' WHERE token = $1",
token
)
.execute(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(())
}
async fn get_reserved_signing_key_full(
&self,
public_key_did_key: &str,
) -> Result<Option<ReservedSigningKeyFull>, DbError> {
let row = sqlx::query!(
r#"SELECT id, did, public_key_did_key, private_key_bytes, expires_at, used_at
FROM reserved_signing_keys WHERE public_key_did_key = $1"#,
public_key_did_key
)
.fetch_optional(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(row.map(|r| ReservedSigningKeyFull {
id: r.id,
did: r.did.map(|d| Did::new(d).expect("valid DID in database")),
public_key_did_key: r.public_key_did_key,
private_key_bytes: r.private_key_bytes,
expires_at: r.expires_at,
used_at: r.used_at,
}))
}
async fn get_plc_tokens_by_did(&self, did: &Did) -> Result<Vec<PlcTokenInfo>, DbError> {
let results = sqlx::query!(
r#"SELECT t.token, t.expires_at
FROM plc_operation_tokens t
JOIN users u ON t.user_id = u.id
WHERE u.did = $1"#,
did.as_str()
)
.fetch_all(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(results
.into_iter()
.map(|r| PlcTokenInfo {
token: r.token,
expires_at: r.expires_at,
})
.collect())
}
async fn count_plc_tokens_by_did(&self, did: &Did) -> Result<i64, DbError> {
let count = sqlx::query_scalar!(
r#"SELECT COUNT(*) as "count!"
FROM plc_operation_tokens t
JOIN users u ON t.user_id = u.id
WHERE u.did = $1"#,
did.as_str()
)
.fetch_one(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(count)
}
}
+2 -2
View File
@@ -28,7 +28,7 @@ use tranquil_db_traits::{
pub use user::PostgresUserRepository;
pub struct PostgresRepositories {
pub pool: PgPool,
pub pool: Option<PgPool>,
pub user: Arc<dyn UserRepository>,
pub oauth: Arc<dyn OAuthRepository>,
pub session: Arc<dyn SessionRepository>,
@@ -44,7 +44,7 @@ pub struct PostgresRepositories {
impl PostgresRepositories {
pub fn new(pool: PgPool) -> Self {
Self {
pool: pool.clone(),
pool: Some(pool.clone()),
user: Arc::new(PostgresUserRepository::new(pool.clone())),
oauth: Arc::new(PostgresOAuthRepository::new(pool.clone())),
session: Arc::new(PostgresSessionRepository::new(pool.clone())),
+15
View File
@@ -1323,4 +1323,19 @@ impl OAuthRepository for PostgresOAuthRepository {
.map_err(map_sqlx_error)?;
Ok(result.rows_affected())
}
async fn get_2fa_challenge_code(
&self,
request_uri: &RequestId,
) -> Result<Option<String>, DbError> {
let code = sqlx::query_scalar!(
"SELECT code FROM oauth_2fa_challenge WHERE request_uri = $1",
request_uri.as_str()
)
.fetch_optional(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(code)
}
}
+78 -2
View File
@@ -234,6 +234,8 @@ impl UserRepository for PostgresUserRepository {
limit: i64,
) -> Result<Vec<AccountSearchResult>, DbError> {
let cursor_str = cursor_did.map(|d| d.as_str());
let email_like = email_filter.map(|e| format!("%{e}%"));
let handle_like = handle_filter.map(|h| format!("%{h}%"));
let rows = sqlx::query!(
r#"SELECT did, handle, email, created_at, email_verified, deactivated_at, invites_disabled
FROM users
@@ -243,8 +245,8 @@ impl UserRepository for PostgresUserRepository {
ORDER BY did ASC
LIMIT $4"#,
cursor_str,
email_filter,
handle_filter,
email_like.as_deref(),
handle_like.as_deref(),
limit
)
.fetch_all(&self.pool)
@@ -627,6 +629,18 @@ impl UserRepository for PostgresUserRepository {
Ok(result.rows_affected())
}
async fn set_admin_status(&self, did: &Did, is_admin: bool) -> Result<(), DbError> {
sqlx::query!(
"UPDATE users SET is_admin = $1 WHERE did = $2",
is_admin,
did.as_str()
)
.execute(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(())
}
async fn get_notification_prefs(
&self,
did: &Did,
@@ -3306,4 +3320,66 @@ impl UserRepository for PostgresUserRepository {
.map_err(map_sqlx_error)?;
Ok(row.flatten())
}
async fn get_password_reset_info(
&self,
email: &str,
) -> Result<Option<tranquil_db_traits::PasswordResetInfo>, DbError> {
let row = sqlx::query!(
"SELECT password_reset_code, password_reset_code_expires_at FROM users WHERE email = $1",
email
)
.fetch_optional(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(row.map(|r| tranquil_db_traits::PasswordResetInfo {
code: r.password_reset_code,
expires_at: r.password_reset_code_expires_at,
}))
}
async fn enable_totp_verified(
&self,
did: &Did,
encrypted_secret: &[u8],
) -> Result<(), DbError> {
sqlx::query!(
r#"INSERT INTO user_totp (did, secret_encrypted, encryption_version, verified, created_at)
VALUES ($1, $2, 1, TRUE, NOW())
ON CONFLICT (did) DO UPDATE SET secret_encrypted = $2, verified = TRUE"#,
did.as_str(),
encrypted_secret
)
.execute(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(())
}
async fn set_two_factor_enabled(&self, did: &Did, enabled: bool) -> Result<(), DbError> {
sqlx::query!(
"UPDATE users SET two_factor_enabled = $1 WHERE did = $2",
enabled,
did.as_str()
)
.execute(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(())
}
async fn expire_password_reset_code(&self, email: &str) -> Result<(), DbError> {
sqlx::query!(
"UPDATE users SET password_reset_code_expires_at = NOW() - INTERVAL '1 hour' WHERE email = $1",
email
)
.execute(&self.pool)
.await
.map_err(map_sqlx_error)?;
Ok(())
}
}
+1
View File
@@ -99,4 +99,5 @@ tranquil-ripple = { workspace = true }
tranquil-sync = { workspace = true }
tranquil-api = { workspace = true }
tranquil-oauth-server = { workspace = true }
tracing-subscriber = { workspace = true, features = ["env-filter"] }
wiremock = { workspace = true }
+136 -70
View File
@@ -48,6 +48,7 @@ pub struct AppState {
pub shutdown: CancellationToken,
pub bootstrap_invite_code: Option<String>,
pub signal_sender: Option<Arc<tranquil_signal::SignalSlot>>,
pub signal_store_provider: Option<Arc<dyn tranquil_signal::SignalStoreProvider>>,
}
#[derive(Debug, Clone, Copy)]
@@ -210,76 +211,122 @@ impl AppState {
pub async fn new(shutdown: CancellationToken) -> Result<Self, Box<dyn Error>> {
let cfg = tranquil_config::get();
let database_url = &cfg.database.url;
let max_connections = cfg.database.max_connections;
let min_connections = cfg.database.min_connections;
let acquire_timeout_secs = cfg.database.acquire_timeout_secs;
tracing::info!(
"Configuring database pool: max={}, min={}, acquire_timeout={}s",
max_connections,
min_connections,
acquire_timeout_secs
);
let db = sqlx::postgres::PgPoolOptions::new()
.max_connections(max_connections)
.min_connections(min_connections)
.acquire_timeout(std::time::Duration::from_secs(acquire_timeout_secs))
.idle_timeout(std::time::Duration::from_secs(300))
.max_lifetime(std::time::Duration::from_secs(1800))
.connect(database_url)
.await
.map_err(|e| format!("Failed to connect to Postgres: {}", e))?;
sqlx::migrate!("./migrations")
.run(&db)
.await
.map_err(|e| format!("Failed to run migrations: {}", e))?;
let bootstrap_invite_code = match (
cfg.server.invite_code_required,
sqlx::query_scalar!("SELECT COUNT(*) FROM users")
.fetch_one(&db)
.await,
) {
(true, Ok(Some(0))) => {
let code = crate::util::gen_invite_code();
match cfg.storage.repo_backend() {
tranquil_config::RepoBackend::TranquilStore => {
tracing::info!(
"No users exist and invite codes are required. Bootstrap invite code: {}",
code
"tranquil-store repo backend active. EXPERIMENTAL! No garbage collection, no backup/restore"
);
Some(code)
Ok(Self::from_store(shutdown).await)
}
_ => None,
};
tranquil_config::RepoBackend::Postgres => {
let database_url = &cfg.database.url;
let max_connections = cfg.database.max_connections;
let min_connections = cfg.database.min_connections;
let acquire_timeout_secs = cfg.database.acquire_timeout_secs;
let mut state = Self::from_db(db, shutdown).await;
state.bootstrap_invite_code = bootstrap_invite_code;
Ok(state)
tracing::info!(
"Configuring database pool: max={}, min={}, acquire_timeout={}s",
max_connections,
min_connections,
acquire_timeout_secs
);
let db = sqlx::postgres::PgPoolOptions::new()
.max_connections(max_connections)
.min_connections(min_connections)
.acquire_timeout(std::time::Duration::from_secs(acquire_timeout_secs))
.idle_timeout(std::time::Duration::from_secs(300))
.max_lifetime(std::time::Duration::from_secs(1800))
.connect(database_url)
.await
.map_err(|e| format!("Failed to connect to Postgres: {}", e))?;
sqlx::migrate!("./migrations")
.run(&db)
.await
.map_err(|e| format!("Failed to run migrations: {}", e))?;
let bootstrap_invite_code = match (
cfg.server.invite_code_required,
sqlx::query_scalar!("SELECT COUNT(*) FROM users")
.fetch_one(&db)
.await,
) {
(true, Ok(Some(0))) => {
let code = crate::util::gen_invite_code();
tracing::info!(
"No users exist and invite codes are required. Bootstrap invite code: {}",
code
);
Some(code)
}
_ => None,
};
let mut state = Self::from_db(db, shutdown).await;
state.bootstrap_invite_code = bootstrap_invite_code;
Ok(state)
}
}
}
pub async fn from_db(db: PgPool, shutdown: CancellationToken) -> Self {
let cfg = tranquil_config::get();
let (repos, block_store, signal_store_provider): (
PostgresRepositories,
crate::repo::AnyBlockStore,
Option<Arc<dyn tranquil_signal::SignalStoreProvider>>,
) = match cfg.storage.repo_backend() == tranquil_config::RepoBackend::TranquilStore {
true => {
let wiring = wire_tranquil_store(&cfg.tranquil_store, shutdown.clone());
(
wiring.repos,
crate::repo::AnyBlockStore::TranquilStore(wiring.blockstore),
Some(wiring.signal_provider),
)
}
false => {
let repos = PostgresRepositories::new(db.clone());
let provider: Arc<dyn tranquil_signal::SignalStoreProvider> =
Arc::new(tranquil_signal::PgSignalStoreProvider { pool: db.clone() });
(
repos,
crate::repo::AnyBlockStore::Postgres(PostgresBlockStore::new(db)),
Some(provider),
)
}
};
Self::build(repos, block_store, signal_store_provider, shutdown).await
}
pub async fn from_store(shutdown: CancellationToken) -> Self {
let cfg = tranquil_config::get();
let wiring = wire_tranquil_store(&cfg.tranquil_store, shutdown.clone());
Self::build(
wiring.repos,
crate::repo::AnyBlockStore::TranquilStore(wiring.blockstore),
Some(wiring.signal_provider),
shutdown,
)
.await
}
async fn build(
repos: PostgresRepositories,
block_store: crate::repo::AnyBlockStore,
signal_store_provider: Option<Arc<dyn tranquil_signal::SignalStoreProvider>>,
shutdown: CancellationToken,
) -> Self {
AuthConfig::init();
init_rate_limit_override();
let mut repos = PostgresRepositories::new(db.clone());
let cfg = tranquil_config::get();
let block_store =
match cfg.storage.repo_backend() == tranquil_config::RepoBackend::TranquilStore {
true => {
let bs = wire_tranquil_store(&mut repos, &cfg.tranquil_store, shutdown.clone());
crate::repo::AnyBlockStore::TranquilStore(bs)
}
false => crate::repo::AnyBlockStore::Postgres(PostgresBlockStore::new(db)),
};
let repos = Arc::new(repos);
let blob_store = create_blob_storage().await;
let firehose_buffer_size = tranquil_config::get().firehose.buffer_size;
let firehose_buffer_size = cfg.firehose.buffer_size;
let (firehose_tx, _) = broadcast::channel(firehose_buffer_size);
let rate_limiters = Arc::new(RateLimiters::new());
let repo_write_locks = Arc::new(RepoWriteLocks::new());
@@ -290,7 +337,7 @@ impl AppState {
let sso_config = SsoConfig::init();
let sso_manager = SsoManager::from_config(sso_config);
let webauthn_config = Arc::new(
WebAuthnConfig::new(&tranquil_config::get().server.hostname)
WebAuthnConfig::new(&cfg.server.hostname)
.expect("Failed to create WebAuthn config at startup"),
);
@@ -311,6 +358,7 @@ impl AppState {
shutdown,
bootstrap_invite_code: None,
signal_sender: None,
signal_store_provider,
}
}
@@ -390,11 +438,16 @@ impl AppState {
}
}
struct TranquilStoreWiring {
blockstore: tranquil_store::blockstore::TranquilBlockStore,
signal_provider: Arc<dyn tranquil_signal::SignalStoreProvider>,
repos: PostgresRepositories,
}
fn wire_tranquil_store(
repos: &mut PostgresRepositories,
store_cfg: &tranquil_config::TranquilStoreConfig,
shutdown: CancellationToken,
) -> tranquil_store::blockstore::TranquilBlockStore {
) -> TranquilStoreWiring {
use tranquil_store::RealIO;
use tranquil_store::blockstore::{BlockStoreConfig, TranquilBlockStore};
use tranquil_store::eventlog::{EventLog, EventLogBridge, EventLogConfig};
@@ -460,6 +513,8 @@ fn wire_tranquil_store(
}
let notifier = bridge.notifier();
let signal_db = metastore.database().clone();
let signal_ks = metastore.signal_keyspace();
let pool = Arc::new(HandlerPool::spawn::<RealIO>(
metastore,
@@ -480,16 +535,27 @@ fn wire_tranquil_store(
tracing::info!(data_dir = %store_cfg.data_dir, "tranquil-store data directory");
repos.repo = Arc::new(client.clone());
repos.backlink = Arc::new(client.clone());
repos.blob = Arc::new(client.clone());
repos.user = Arc::new(client.clone());
repos.session = Arc::new(client.clone());
repos.oauth = Arc::new(client.clone());
repos.infra = Arc::new(client.clone());
repos.delegation = Arc::new(client.clone());
repos.sso = Arc::new(client);
repos.event_notifier = Arc::new(notifier);
let repos = PostgresRepositories {
pool: None,
repo: Arc::new(client.clone()),
backlink: Arc::new(client.clone()),
blob: Arc::new(client.clone()),
user: Arc::new(client.clone()),
session: Arc::new(client.clone()),
oauth: Arc::new(client.clone()),
infra: Arc::new(client.clone()),
delegation: Arc::new(client.clone()),
sso: Arc::new(client),
event_notifier: Arc::new(notifier),
};
blockstore
let signal_provider: Arc<dyn tranquil_signal::SignalStoreProvider> = Arc::new(
tranquil_signal::fjall_store::FjallSignalStoreProvider::new(signal_db, signal_ks),
);
TranquilStoreWiring {
blockstore,
signal_provider,
repos,
}
}
@@ -1,32 +1,37 @@
mod common;
use common::{base_url, client, create_account_and_login, get_test_db_pool};
use common::{base_url, client, create_account_and_login, get_test_repos};
use serde_json::{Value, json};
use tranquil_db_traits::{CommsChannel, CommsType};
use tranquil_types::Did;
#[tokio::test]
async fn test_get_notification_history() {
let client = client();
let base = base_url().await;
let pool = get_test_db_pool().await;
let repos = get_test_repos().await;
let (token, did) = create_account_and_login(&client).await;
let user_id: uuid::Uuid = sqlx::query_scalar("SELECT id FROM users WHERE did = $1")
.bind(&did)
.fetch_one(pool)
let user_id = repos
.user
.get_id_by_did(&Did::new(did).unwrap())
.await
.expect("DB error")
.expect("User not found");
for i in 0..3 {
sqlx::query(
r#"INSERT INTO comms_queue (user_id, channel, comms_type, recipient, subject, body)
VALUES ($1, 'email', 'welcome', $2, $3, $4)"#,
)
.bind(user_id)
.bind("test@example.com")
.bind(format!("Subject {}", i))
.bind(format!("Body {}", i))
.execute(pool)
.await
.expect("Failed to enqueue");
repos
.infra
.enqueue_comms(
Some(user_id),
CommsChannel::Email,
CommsType::Welcome,
"test@example.com",
Some(&format!("Subject {}", i)),
&format!("Body {}", i),
None,
)
.await
.expect("Failed to enqueue");
}
let resp = client
@@ -140,7 +145,7 @@ async fn test_verify_channel_not_set() {
async fn test_update_email_via_notification_prefs() {
let client = client();
let base = base_url().await;
let pool = get_test_db_pool().await;
let repos = get_test_repos().await;
let (token, did) = create_account_and_login(&client).await;
let unique_email = format!("newemail_{}@example.com", uuid::Uuid::new_v4());
@@ -163,19 +168,22 @@ async fn test_update_email_via_notification_prefs() {
.contains(&json!("email"))
);
let user_id: uuid::Uuid = sqlx::query_scalar("SELECT id FROM users WHERE did = $1")
.bind(&did)
.fetch_one(pool)
let user_id = repos
.user
.get_id_by_did(&Did::new(did).unwrap())
.await
.expect("DB error")
.expect("User not found");
let body_text: String = sqlx::query_scalar(
"SELECT body FROM comms_queue WHERE user_id = $1 AND comms_type = 'email_update' ORDER BY created_at DESC LIMIT 1",
)
.bind(user_id)
.fetch_one(pool)
.await
.expect("Verification code not found");
let comms = repos
.infra
.get_latest_comms_for_user(user_id, CommsType::EmailUpdate, 1)
.await
.expect("DB error");
let body_text = comms
.first()
.map(|c| c.body.clone())
.expect("Verification code not found");
let code = body_text
.lines()
+34 -21
View File
@@ -2,12 +2,14 @@ mod common;
use reqwest::StatusCode;
use serde_json::{Value, json};
use tranquil_db_traits::CommsType;
use tranquil_types::Did;
#[tokio::test]
async fn test_send_email_success() {
let client = common::client();
let base_url = common::base_url().await;
let pool = common::get_test_db_pool().await;
let repos = common::get_test_repos().await;
let (access_jwt, did) = common::create_admin_account_and_login(&client).await;
let res = client
.post(format!("{}/xrpc/com.atproto.admin.sendEmail", base_url))
@@ -24,17 +26,18 @@ async fn test_send_email_success() {
assert_eq!(res.status(), StatusCode::OK);
let body: Value = res.json().await.expect("Invalid JSON");
assert_eq!(body["sent"], true);
let user = sqlx::query!("SELECT id FROM users WHERE did = $1", did)
.fetch_one(pool)
let user_id = repos
.user
.get_id_by_did(&Did::new(did).unwrap())
.await
.expect("DB error")
.expect("User not found");
let notification = sqlx::query!(
"SELECT subject, body, comms_type as \"comms_type: String\" FROM comms_queue WHERE user_id = $1 AND comms_type = 'admin_email' ORDER BY created_at DESC LIMIT 1",
user.id
)
.fetch_one(pool)
.await
.expect("Notification not found");
let comms = repos
.infra
.get_latest_comms_for_user(user_id, CommsType::AdminEmail, 1)
.await
.expect("DB error");
let notification = comms.first().expect("Notification not found");
assert_eq!(notification.subject.as_deref(), Some("Test Admin Email"));
assert!(
notification
@@ -47,7 +50,7 @@ async fn test_send_email_success() {
async fn test_send_email_default_subject() {
let client = common::client();
let base_url = common::base_url().await;
let pool = common::get_test_db_pool().await;
let repos = common::get_test_repos().await;
let (access_jwt, did) = common::create_admin_account_and_login(&client).await;
let res = client
.post(format!("{}/xrpc/com.atproto.admin.sendEmail", base_url))
@@ -63,19 +66,29 @@ async fn test_send_email_default_subject() {
assert_eq!(res.status(), StatusCode::OK);
let body: Value = res.json().await.expect("Invalid JSON");
assert_eq!(body["sent"], true);
let user = sqlx::query!("SELECT id FROM users WHERE did = $1", did)
.fetch_one(pool)
let user_id = repos
.user
.get_id_by_did(&Did::new(did).unwrap())
.await
.expect("DB error")
.expect("User not found");
let notification = sqlx::query!(
"SELECT subject FROM comms_queue WHERE user_id = $1 AND comms_type = 'admin_email' AND body = 'Email without subject' LIMIT 1",
user.id
)
.fetch_one(pool)
.await
.expect("Notification not found");
let comms = repos
.infra
.get_latest_comms_for_user(user_id, CommsType::AdminEmail, 10)
.await
.expect("DB error");
let notification = comms
.iter()
.find(|c| c.body == "Email without subject")
.expect("Notification not found");
assert!(notification.subject.is_some());
assert!(notification.subject.unwrap().contains("Message from"));
assert!(
notification
.subject
.as_ref()
.unwrap()
.contains("Message from")
);
}
#[tokio::test]
+4 -3
View File
@@ -215,9 +215,10 @@ async fn test_oauth_admin_extractor_allows_oauth_tokens() {
let did = account["did"].as_str().unwrap().to_string();
verify_new_account(&http_client, &did).await;
let pool = common::get_test_db_pool().await;
sqlx::query!("UPDATE users SET is_admin = TRUE WHERE did = $1", &did)
.execute(pool)
let repos = common::get_test_repos().await;
repos
.user
.set_admin_status(&tranquil_types::Did::new(did.clone()).unwrap(), true)
.await
.expect("Failed to mark user as admin");
+105 -43
View File
@@ -28,10 +28,18 @@ static MOCK_PLC: OnceLock<MockServer> = OnceLock::new();
static TEST_DB_POOL: OnceLock<sqlx::PgPool> = OnceLock::new();
static TEST_TEMP_DIR: OnceLock<PathBuf> = OnceLock::new();
static CLUSTER: OnceLock<Vec<ServerInstance>> = OnceLock::new();
static TEST_REPOS: OnceLock<Arc<tranquil_db::PostgresRepositories>> = OnceLock::new();
#[allow(dead_code)]
pub fn is_store_backend() -> bool {
std::env::var("TRANQUIL_TEST_BACKEND")
.map(|v| v == "store")
.unwrap_or(false)
}
#[allow(dead_code)]
pub struct ServerConfig {
pub pool: sqlx::PgPool,
pub pool: Option<sqlx::PgPool>,
pub cache: Option<(Arc<dyn Cache>, Arc<dyn DistributedRateLimiter>)>,
}
@@ -123,6 +131,12 @@ pub async fn base_url() -> &'static str {
SERVER_URL.get_or_init(|| {
let (tx, rx) = std::sync::mpsc::channel();
std::thread::spawn(move || {
let _ = tracing_subscriber::fmt()
.with_env_filter(
tracing_subscriber::EnvFilter::try_from_default_env()
.unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("warn")),
)
.try_init();
unsafe {
std::env::set_var("TRANQUIL_PDS_ALLOW_INSECURE_SECRETS", "1");
}
@@ -141,7 +155,10 @@ pub async fn base_url() -> &'static str {
}
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async move {
if has_external_infra() {
if is_store_backend() {
let url = setup_store_backend().await;
tx.send(url).unwrap();
} else if has_external_infra() {
let url = setup_with_external_infra().await;
tx.send(url).unwrap();
} else {
@@ -557,9 +574,12 @@ async fn spawn_server(config: ServerConfig) -> ServerInstance {
.with_oauth_authorize_limit(10000)
.with_oauth_token_limit(10000);
let cache_refs = config.cache.as_ref().map(|(c, r)| (c.clone(), r.clone()));
let mut state = AppState::from_db(config.pool, CancellationToken::new())
.await
.with_rate_limiters(rate_limiters);
let mut state = match config.pool {
Some(pool) => AppState::from_db(pool, CancellationToken::new()).await,
None => AppState::from_store(CancellationToken::new()).await,
};
state = state.with_rate_limiters(rate_limiters);
TEST_REPOS.set(state.repos.clone()).ok();
if let Some((cache, distributed_rate_limiter)) = config.cache {
state = state.with_cache(cache, distributed_rate_limiter);
}
@@ -590,6 +610,39 @@ async fn spawn_server(config: ServerConfig) -> ServerInstance {
}
}
async fn setup_store_backend() -> String {
let temp_dir =
std::env::temp_dir().join(format!("tranquil-pds-store-{}", uuid::Uuid::new_v4()));
let blob_path = temp_dir.join("blobs");
let backup_path = temp_dir.join("backups");
let store_path = temp_dir.join("store");
std::fs::create_dir_all(&blob_path).expect("failed to create blob temp directory");
std::fs::create_dir_all(&backup_path).expect("failed to create backup temp directory");
std::fs::create_dir_all(&store_path).expect("failed to create store temp directory");
TEST_TEMP_DIR.set(temp_dir).ok();
let plc_url = setup_mock_plc_directory().await;
unsafe {
std::env::set_var("BLOB_STORAGE_BACKEND", "filesystem");
std::env::set_var("BLOB_STORAGE_PATH", blob_path.to_str().unwrap());
std::env::set_var("BACKUP_STORAGE_BACKEND", "filesystem");
std::env::set_var("BACKUP_STORAGE_PATH", backup_path.to_str().unwrap());
std::env::set_var("MAX_IMPORT_SIZE", "100000000");
std::env::set_var("SKIP_IMPORT_VERIFICATION", "true");
std::env::set_var("PLC_DIRECTORY_URL", &plc_url);
std::env::set_var("REPO_BACKEND", "tranquil-store");
std::env::set_var("TRANQUIL_STORE_DATA_DIR", store_path.to_str().unwrap());
std::env::set_var("DATABASE_URL", "postgres://unused/unused");
}
register_mock_appview().await;
let instance = spawn_server(ServerConfig {
pool: None,
cache: None,
})
.await;
APP_PORT.set(instance.port).ok();
instance.url
}
async fn spawn_app(database_url: String) -> String {
let pool = PgPoolOptions::new()
.max_connections(10)
@@ -608,7 +661,11 @@ async fn spawn_app(database_url: String) -> String {
.await
.expect("Failed to create test pool");
TEST_DB_POOL.set(test_pool).ok();
let instance = spawn_server(ServerConfig { pool, cache: None }).await;
let instance = spawn_server(ServerConfig {
pool: Some(pool),
cache: None,
})
.await;
APP_PORT.set(instance.port).ok();
instance.url
}
@@ -659,7 +716,7 @@ pub async fn spawn_cluster(database_url: String, node_count: usize) -> Vec<Serve
let mut instances: Vec<ServerInstance> = Vec::with_capacity(node_count);
for (cache, rate_limiter) in ripple_nodes {
let server_config = ServerConfig {
pool: pool.clone(),
pool: Some(pool.clone()),
cache: Some((cache, rate_limiter)),
};
let instance = spawn_server(server_config).await;
@@ -799,18 +856,14 @@ pub async fn get_test_db_pool() -> &'static sqlx::PgPool {
}
#[allow(dead_code)]
pub async fn verify_new_account(client: &Client, did: &str) -> String {
let pool = get_test_db_pool().await;
let body_text: String = sqlx::query_scalar!(
"SELECT body FROM comms_queue WHERE user_id = (SELECT id FROM users WHERE did = $1) AND comms_type = 'email_verification' ORDER BY created_at DESC LIMIT 1",
did
)
.fetch_one(pool)
.await
.expect("Failed to get verification code");
pub async fn get_test_repos() -> &'static Arc<tranquil_db::PostgresRepositories> {
base_url().await;
TEST_REPOS.get().expect("TEST_REPOS not initialized")
}
fn extract_verification_code(body_text: &str) -> String {
let lines: Vec<&str> = body_text.lines().collect();
let verification_code = lines
lines
.iter()
.enumerate()
.find(|(_, line)| line.contains("verification code is:") || line.contains("code is:"))
@@ -821,7 +874,35 @@ pub async fn verify_new_account(client: &Client, did: &str) -> String {
.find(|line| line.trim().starts_with("MX"))
.map(|s| s.trim().to_string())
})
.unwrap_or_else(|| body_text.clone());
.unwrap_or_else(|| body_text.to_string())
}
async fn get_verification_body_for_did(did: &str) -> String {
use tranquil_db_traits::CommsType;
use tranquil_types::Did;
let repos = get_test_repos().await;
let user = repos
.user
.get_by_did(&Did::new(did.to_string()).unwrap())
.await
.expect("failed to look up user")
.expect("user not found");
let comms = repos
.infra
.get_latest_comms_for_user(user.id, CommsType::EmailVerification, 1)
.await
.expect("failed to get comms");
comms
.first()
.map(|c| c.body.clone())
.expect("no email_verification comms found")
}
#[allow(dead_code)]
pub async fn verify_new_account(client: &Client, did: &str) -> String {
let body_text = get_verification_body_for_did(did).await;
let verification_code = extract_verification_code(&body_text);
let confirm_payload = json!({
"did": did,
@@ -956,10 +1037,11 @@ async fn create_account_and_login_internal(client: &Client, make_admin: bool) ->
if res.status() == StatusCode::OK {
let body: Value = res.json().await.expect("Invalid JSON");
let did = body["did"].as_str().expect("No did").to_string();
let pool = get_test_db_pool().await;
if make_admin {
sqlx::query!("UPDATE users SET is_admin = TRUE WHERE did = $1", &did)
.execute(pool)
let repos = get_test_repos().await;
repos
.user
.set_admin_status(&tranquil_types::Did::new(did.clone()).unwrap(), true)
.await
.expect("Failed to mark user as admin");
}
@@ -969,28 +1051,8 @@ async fn create_account_and_login_internal(client: &Client, make_admin: bool) ->
{
return (access_jwt.to_string(), did);
}
let body_text: String = sqlx::query_scalar!(
"SELECT body FROM comms_queue WHERE user_id = (SELECT id FROM users WHERE did = $1) AND comms_type = 'email_verification' ORDER BY created_at DESC LIMIT 1",
&did
)
.fetch_one(pool)
.await
.expect("Failed to get verification from comms_queue");
let lines: Vec<&str> = body_text.lines().collect();
let verification_code = lines
.iter()
.enumerate()
.find(|(_, line): &(usize, &&str)| {
line.contains("verification code is:") || line.contains("code is:")
})
.and_then(|(i, _)| lines.get(i + 1).map(|s: &&str| s.trim().to_string()))
.or_else(|| {
body_text
.lines()
.find(|line| line.trim().starts_with("MX"))
.map(|s| s.trim().to_string())
})
.unwrap_or_else(|| body_text.clone());
let body_text = get_verification_body_for_did(&did).await;
let verification_code = extract_verification_code(&body_text);
let confirm_payload = json!({
"did": did,
+51 -60
View File
@@ -51,15 +51,14 @@ async fn test_delete_account_full_flow() {
.await
.expect("Failed to request account deletion");
assert_eq!(request_delete_res.status(), StatusCode::OK);
let pool = get_test_db_pool().await;
let row = sqlx::query!(
"SELECT token FROM account_deletion_requests WHERE did = $1",
did
)
.fetch_one(pool)
.await
.expect("Failed to query deletion token");
let token = row.token;
let repos = get_test_repos().await;
let deletion_request = repos
.infra
.get_deletion_request_by_did(&tranquil_types::Did::new(did.clone()).unwrap())
.await
.unwrap()
.unwrap();
let token = deletion_request.token;
let delete_payload = json!({
"did": did,
"password": password,
@@ -75,11 +74,12 @@ async fn test_delete_account_full_flow() {
.await
.expect("Failed to delete account");
assert_eq!(delete_res.status(), StatusCode::OK);
let user_row = sqlx::query!("SELECT id FROM users WHERE did = $1", did)
.fetch_optional(pool)
let user = repos
.user
.get_by_did(&tranquil_types::Did::new(did.clone()).unwrap())
.await
.expect("Failed to query user");
assert!(user_row.is_none(), "User should be deleted from database");
.unwrap();
assert!(user.is_none(), "User should be deleted from database");
let session_res = client
.get(format!("{}/xrpc/com.atproto.server.getSession", base_url))
.bearer_auth(&jwt)
@@ -108,15 +108,14 @@ async fn test_delete_account_wrong_password() {
.await
.expect("Failed to request account deletion");
assert_eq!(request_delete_res.status(), StatusCode::OK);
let pool = get_test_db_pool().await;
let row = sqlx::query!(
"SELECT token FROM account_deletion_requests WHERE did = $1",
did
)
.fetch_one(pool)
.await
.expect("Failed to query deletion token");
let token = row.token;
let repos = get_test_repos().await;
let deletion_request = repos
.infra
.get_deletion_request_by_did(&tranquil_types::Did::new(did.clone()).unwrap())
.await
.unwrap()
.unwrap();
let token = deletion_request.token;
let delete_payload = json!({
"did": did,
"password": "wrong-password",
@@ -198,22 +197,15 @@ async fn test_delete_account_expired_token() {
.await
.expect("Failed to request account deletion");
assert_eq!(request_delete_res.status(), StatusCode::OK);
let pool = get_test_db_pool().await;
let row = sqlx::query!(
"SELECT token FROM account_deletion_requests WHERE did = $1",
did
)
.fetch_one(pool)
.await
.expect("Failed to query deletion token");
let token = row.token;
sqlx::query!(
"UPDATE account_deletion_requests SET expires_at = NOW() - INTERVAL '1 hour' WHERE token = $1",
token
)
.execute(pool)
.await
.expect("Failed to expire token");
let repos = get_test_repos().await;
let deletion_request = repos
.infra
.get_deletion_request_by_did(&tranquil_types::Did::new(did.clone()).unwrap())
.await
.unwrap()
.unwrap();
let token = deletion_request.token;
repos.infra.expire_deletion_request(&token).await.unwrap();
let delete_payload = json!({
"did": did,
"password": password,
@@ -257,15 +249,14 @@ async fn test_delete_account_token_mismatch() {
.await
.expect("Failed to request account deletion");
assert_eq!(request_delete_res.status(), StatusCode::OK);
let pool = get_test_db_pool().await;
let row = sqlx::query!(
"SELECT token FROM account_deletion_requests WHERE did = $1",
did1
)
.fetch_one(pool)
.await
.expect("Failed to query deletion token");
let token = row.token;
let repos = get_test_repos().await;
let deletion_request = repos
.infra
.get_deletion_request_by_did(&tranquil_types::Did::new(did1.clone()).unwrap())
.await
.unwrap()
.unwrap();
let token = deletion_request.token;
let delete_payload = json!({
"did": did2,
"password": password2,
@@ -318,15 +309,14 @@ async fn test_delete_account_with_app_password() {
.await
.expect("Failed to request account deletion");
assert_eq!(request_delete_res.status(), StatusCode::OK);
let pool = get_test_db_pool().await;
let row = sqlx::query!(
"SELECT token FROM account_deletion_requests WHERE did = $1",
did
)
.fetch_one(pool)
.await
.expect("Failed to query deletion token");
let token = row.token;
let repos = get_test_repos().await;
let deletion_request = repos
.infra
.get_deletion_request_by_did(&tranquil_types::Did::new(did.clone()).unwrap())
.await
.unwrap()
.unwrap();
let token = deletion_request.token;
let delete_payload = json!({
"did": did,
"password": app_password,
@@ -342,11 +332,12 @@ async fn test_delete_account_with_app_password() {
.await
.expect("Failed to delete account");
assert_eq!(delete_res.status(), StatusCode::OK);
let user_row = sqlx::query!("SELECT id FROM users WHERE did = $1", did)
.fetch_optional(pool)
let user = repos
.user
.get_by_did(&tranquil_types::Did::new(did.clone()).unwrap())
.await
.expect("Failed to query user");
assert!(user_row.is_none(), "User should be deleted from database");
.unwrap();
assert!(user.is_none(), "User should be deleted from database");
}
#[tokio::test]
+89 -50
View File
@@ -1,16 +1,24 @@
mod common;
use reqwest::StatusCode;
use serde_json::{Value, json};
use sqlx::PgPool;
use tranquil_db_traits::CommsType;
use tranquil_types::Did;
async fn get_email_update_token(pool: &PgPool, did: &str) -> String {
let body_text: String = sqlx::query_scalar!(
"SELECT body FROM comms_queue WHERE user_id = (SELECT id FROM users WHERE did = $1) AND comms_type = 'email_update' ORDER BY created_at DESC LIMIT 1",
did
)
.fetch_one(pool)
.await
.expect("Verification not found");
async fn get_email_update_token(did: &str) -> String {
let repos = common::get_test_repos().await;
let parsed_did = Did::new(did.to_string()).unwrap();
let user = repos
.user
.get_by_did(&parsed_did)
.await
.expect("failed to look up user")
.expect("user not found");
let comms = repos
.infra
.get_latest_comms_for_user(user.id, CommsType::EmailUpdate, 1)
.await
.expect("failed to get comms");
let body_text = comms.first().expect("Verification not found").body.clone();
body_text
.lines()
@@ -82,7 +90,7 @@ async fn test_request_email_update_returns_token_required() {
async fn test_update_email_flow_success() {
let client = common::client();
let base_url = common::base_url().await;
let pool = common::get_test_db_pool().await;
let repos = common::get_test_repos().await;
let handle = format!("eu{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
let email = format!("{}@example.com", handle);
let (access_jwt, did) = create_verified_account(&client, base_url, &handle, &email).await;
@@ -101,7 +109,7 @@ async fn test_update_email_flow_success() {
let body: Value = res.json().await.expect("Invalid JSON");
assert_eq!(body["tokenRequired"], true);
let code = get_email_update_token(pool, &did).await;
let code = get_email_update_token(&did).await;
let res = client
.post(format!("{}/xrpc/com.atproto.server.updateEmail", base_url))
@@ -115,11 +123,14 @@ async fn test_update_email_flow_success() {
.expect("Failed to update email");
assert_eq!(res.status(), StatusCode::OK);
let user_email: Option<String> =
sqlx::query_scalar!("SELECT email FROM users WHERE did = $1", did)
.fetch_one(pool)
.await
.expect("User not found");
let parsed_did = Did::new(did).unwrap();
let user_email = repos
.user
.get_email_info_by_did(&parsed_did)
.await
.expect("failed to look up user")
.expect("user not found")
.email;
assert_eq!(user_email, Some(new_email));
}
@@ -239,7 +250,7 @@ async fn test_update_email_invalid_format() {
async fn test_confirm_email_confirms_existing_email() {
let client = common::client();
let base_url = common::base_url().await;
let pool = common::get_test_db_pool().await;
let repos = common::get_test_repos().await;
let handle = format!("ec{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
let email = format!("{}@example.com", handle);
@@ -264,13 +275,23 @@ async fn test_confirm_email_confirms_existing_email() {
.expect("No accessJwt")
.to_string();
let body_text: String = sqlx::query_scalar!(
"SELECT body FROM comms_queue WHERE user_id = (SELECT id FROM users WHERE did = $1) AND comms_type = 'email_verification' ORDER BY created_at DESC LIMIT 1",
did
)
.fetch_one(pool)
.await
.expect("Verification email not found");
let parsed_did = Did::new(did.clone()).unwrap();
let user = repos
.user
.get_by_did(&parsed_did)
.await
.expect("failed to look up user")
.expect("user not found");
let comms = repos
.infra
.get_latest_comms_for_user(user.id, CommsType::EmailVerification, 1)
.await
.expect("failed to get comms");
let body_text = comms
.first()
.expect("Verification email not found")
.body
.clone();
let code = body_text
.lines()
@@ -290,11 +311,13 @@ async fn test_confirm_email_confirms_existing_email() {
.expect("Failed to confirm email");
assert_eq!(res.status(), StatusCode::OK);
let verified: bool =
sqlx::query_scalar!("SELECT email_verified FROM users WHERE did = $1", did)
.fetch_one(pool)
.await
.expect("User not found");
let verified = repos
.user
.get_email_info_by_did(&parsed_did)
.await
.expect("failed to look up user")
.expect("user not found")
.email_verified;
assert!(verified);
}
@@ -302,7 +325,7 @@ async fn test_confirm_email_confirms_existing_email() {
async fn test_confirm_email_rejects_wrong_email() {
let client = common::client();
let base_url = common::base_url().await;
let pool = common::get_test_db_pool().await;
let repos = common::get_test_repos().await;
let handle = format!("ew{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
let email = format!("{}@example.com", handle);
@@ -327,13 +350,23 @@ async fn test_confirm_email_rejects_wrong_email() {
.expect("No accessJwt")
.to_string();
let body_text: String = sqlx::query_scalar!(
"SELECT body FROM comms_queue WHERE user_id = (SELECT id FROM users WHERE did = $1) AND comms_type = 'email_verification' ORDER BY created_at DESC LIMIT 1",
did
)
.fetch_one(pool)
.await
.expect("Verification email not found");
let parsed_did = Did::new(did).unwrap();
let user = repos
.user
.get_by_did(&parsed_did)
.await
.expect("failed to look up user")
.expect("user not found");
let comms = repos
.infra
.get_latest_comms_for_user(user.id, CommsType::EmailVerification, 1)
.await
.expect("failed to get comms");
let body_text = comms
.first()
.expect("Verification email not found")
.body
.clone();
let code = body_text
.lines()
@@ -402,7 +435,7 @@ async fn test_confirm_email_invalid_token() {
async fn test_unverified_account_can_update_email_without_token() {
let client = common::client();
let base_url = common::base_url().await;
let pool = common::get_test_db_pool().await;
let repos = common::get_test_repos().await;
let handle = format!("ev{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
let email = format!("{}@example.com", handle);
@@ -457,11 +490,14 @@ async fn test_unverified_account_can_update_email_without_token() {
"Unverified account should be able to update email without token"
);
let user_email: Option<String> =
sqlx::query_scalar!("SELECT email FROM users WHERE did = $1", did)
.fetch_one(pool)
.await
.expect("User not found");
let parsed_did = Did::new(did).unwrap();
let user_email = repos
.user
.get_email_info_by_did(&parsed_did)
.await
.expect("failed to look up user")
.expect("user not found")
.email;
assert_eq!(user_email, Some(new_email));
}
@@ -469,7 +505,7 @@ async fn test_unverified_account_can_update_email_without_token() {
async fn test_update_email_to_same_as_another_user_allowed() {
let client = common::client();
let base_url = common::base_url().await;
let pool = common::get_test_db_pool().await;
let repos = common::get_test_repos().await;
let handle1 = format!("d1{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
let email1 = format!("{}@example.com", handle1);
@@ -490,7 +526,7 @@ async fn test_update_email_to_same_as_another_user_allowed() {
.expect("Failed to request email update");
assert_eq!(res.status(), StatusCode::OK);
let code = get_email_update_token(pool, &did2).await;
let code = get_email_update_token(&did2).await;
let res = client
.post(format!("{}/xrpc/com.atproto.server.updateEmail", base_url))
@@ -508,10 +544,13 @@ async fn test_update_email_to_same_as_another_user_allowed() {
"Multiple accounts can share the same email address"
);
let user_email: Option<String> =
sqlx::query_scalar!("SELECT email FROM users WHERE did = $1", did2)
.fetch_one(pool)
.await
.expect("User not found");
let parsed_did = Did::new(did2).unwrap();
let user_email = repos
.user
.get_email_info_by_did(&parsed_did)
.await
.expect("failed to look up user")
.expect("user not found")
.email;
assert_eq!(user_email, Some(email1.clone()));
}
@@ -800,11 +800,8 @@ async fn test_firehose_outdated_cursor_info() {
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
let pool = get_test_db_pool().await;
let max_seq: i64 = sqlx::query_scalar::<_, i64>("SELECT COALESCE(MAX(seq), 0) FROM repo_seq")
.fetch_one(pool)
.await
.unwrap();
let repos = get_test_repos().await;
let max_seq = repos.repo.get_max_seq().await.unwrap().as_i64();
let outdated_cursor = (max_seq - 100).max(1);
let url = format!(
"ws://127.0.0.1:{}/xrpc/com.atproto.sync.subscribeRepos?cursor={}",
+7 -15
View File
@@ -482,19 +482,11 @@ pub fn get_multikey_from_signing_key(signing_key: &k256::ecdsa::SigningKey) -> S
#[allow(dead_code)]
pub async fn get_user_signing_key(did: &str) -> Option<Vec<u8>> {
let db_url = get_db_connection_string().await;
let pool = sqlx::PgPool::connect(&db_url).await.ok()?;
let row = sqlx::query!(
r#"
SELECT k.key_bytes, k.encryption_version
FROM user_keys k
JOIN users u ON k.user_id = u.id
WHERE u.did = $1
"#,
did
)
.fetch_optional(&pool)
.await
.ok()??;
tranquil_pds::config::decrypt_key(&row.key_bytes, row.encryption_version).ok()
let repos = super::common::get_test_repos().await;
let key_info = repos
.user
.get_user_key_by_did(&tranquil_types::Did::new(did.to_string()).ok()?)
.await
.ok()??;
tranquil_pds::config::decrypt_key(&key_info.key_bytes, key_info.encryption_version).ok()
}
+1
View File
@@ -203,6 +203,7 @@ async fn test_create_invite_codes_no_auth() {
#[tokio::test]
async fn test_create_invite_codes_non_admin() {
let client = client();
let _ = create_account_and_login(&client).await;
let (access_jwt, _did) = create_account_and_login(&client).await;
let payload = json!({
"useCount": 2
+14 -6
View File
@@ -2,7 +2,7 @@
mod common;
use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
use chrono::{Duration, Utc};
use common::{base_url, client, create_account_and_login, get_test_db_pool};
use common::{base_url, client, create_account_and_login, get_test_repos};
use k256::SecretKey;
use k256::ecdsa::{Signature, SigningKey, signature::Signer};
use rand::rngs::OsRng;
@@ -691,11 +691,19 @@ async fn test_refresh_token_replay_protection() {
let account: Value = create_res.json().await.unwrap();
let did = account["did"].as_str().unwrap();
let pool = get_test_db_pool().await;
let body_text: String = sqlx::query_scalar!(
"SELECT body FROM comms_queue WHERE user_id = (SELECT id FROM users WHERE did = $1) AND comms_type = 'email_verification' ORDER BY created_at DESC LIMIT 1",
did
).fetch_one(pool).await.unwrap();
let repos = get_test_repos().await;
let user = repos
.user
.get_by_did(&tranquil_types::Did::new(did.to_string()).unwrap())
.await
.unwrap()
.unwrap();
let comms = repos
.infra
.get_latest_comms_for_user(user.id, tranquil_db_traits::CommsType::EmailVerification, 1)
.await
.unwrap();
let body_text = comms.first().unwrap().body.clone();
let lines: Vec<&str> = body_text.lines().collect();
let code = lines
.iter()
+79 -112
View File
@@ -1,54 +1,53 @@
mod common;
use common::{base_url, client, create_account_and_login, get_test_db_pool};
use common::{base_url, client, create_account_and_login, get_test_repos};
use reqwest::StatusCode;
use serde_json::{Value, json};
use tranquil_db_traits::CommsType;
use tranquil_types::Did;
async fn enable_totp_for_user(did: &str) {
let pool = get_test_db_pool().await;
let secret = vec![0u8; 20];
sqlx::query(
r#"INSERT INTO user_totp (did, secret_encrypted, encryption_version, verified, created_at)
VALUES ($1, $2, 1, TRUE, NOW())
ON CONFLICT (did) DO UPDATE SET verified = TRUE"#,
)
.bind(did)
.bind(&secret)
.execute(pool)
.await
.expect("Failed to enable TOTP");
let repos = get_test_repos().await;
repos
.user
.enable_totp_verified(&Did::new(did.to_string()).unwrap(), &[0u8; 20])
.await
.unwrap();
}
async fn set_allow_legacy_login(did: &str, allow: bool) {
let pool = get_test_db_pool().await;
sqlx::query("UPDATE users SET allow_legacy_login = $1 WHERE did = $2")
.bind(allow)
.bind(did)
.execute(pool)
let repos = get_test_repos().await;
repos
.user
.update_legacy_login(&Did::new(did.to_string()).unwrap(), allow)
.await
.expect("Failed to set allow_legacy_login");
.unwrap();
}
async fn get_2fa_code_from_queue(did: &str) -> Option<String> {
let pool = get_test_db_pool().await;
let row: Option<(String,)> = sqlx::query_as(
r#"SELECT body FROM comms_queue
WHERE user_id = (SELECT id FROM users WHERE did = $1)
AND comms_type = 'two_factor_code'
ORDER BY created_at DESC LIMIT 1"#,
)
.bind(did)
.fetch_optional(pool)
.await
.ok()
.flatten();
let repos = get_test_repos().await;
let parsed_did = Did::new(did.to_string()).unwrap();
let user_id = repos
.user
.get_id_by_did(&parsed_did)
.await
.expect("DB error")
.expect("User not found");
row.and_then(|(body,)| {
body.lines()
let comms = repos
.infra
.get_latest_comms_for_user(user_id, CommsType::TwoFactorCode, 1)
.await
.ok()?;
comms.first().and_then(|c| {
c.body
.lines()
.find(|line: &&str| line.chars().all(|c: char| c.is_ascii_digit()) && line.len() == 8)
.map(|s: &str| s.to_string())
.or_else(|| {
body.split_whitespace()
c.body
.split_whitespace()
.find(|word: &&str| {
word.chars().all(|c: char| c.is_ascii_digit()) && word.len() == 8
})
@@ -58,39 +57,47 @@ async fn get_2fa_code_from_queue(did: &str) -> Option<String> {
}
async fn clear_2fa_challenges_for_user(did: &str) {
let pool = get_test_db_pool().await;
let _ = sqlx::query(
"DELETE FROM comms_queue WHERE user_id = (SELECT id FROM users WHERE did = $1) AND comms_type = 'two_factor_code'",
)
.bind(did)
.execute(pool)
.await;
let repos = get_test_repos().await;
let parsed_did = Did::new(did.to_string()).unwrap();
let user_id = repos
.user
.get_id_by_did(&parsed_did)
.await
.expect("DB error")
.expect("User not found");
let _ = repos
.infra
.delete_comms_by_type_for_user(user_id, CommsType::TwoFactorCode)
.await;
}
async fn set_email_auth_factor(did: &str, enabled: bool) {
let pool = get_test_db_pool().await;
let user_id: uuid::Uuid =
sqlx::query_scalar::<_, uuid::Uuid>("SELECT id FROM users WHERE did = $1")
.bind(did)
.fetch_one(pool)
.await
.expect("Failed to get user id");
let pool = get_test_db_pool().await;
let _ = sqlx::query(
"DELETE FROM account_preferences WHERE user_id = $1 AND name = 'email_auth_factor'",
)
.bind(user_id)
.execute(pool)
.await;
let pool = get_test_db_pool().await;
sqlx::query(
"INSERT INTO account_preferences (user_id, name, value_json) VALUES ($1, 'email_auth_factor', $2::jsonb)",
)
.bind(user_id)
.bind(serde_json::json!(enabled))
.execute(pool)
.await
.expect("Failed to set email_auth_factor");
let repos = get_test_repos().await;
let parsed_did = Did::new(did.to_string()).unwrap();
let user_id = repos
.user
.get_id_by_did(&parsed_did)
.await
.expect("DB error")
.expect("User not found");
repos
.infra
.upsert_account_preference(user_id, "email_auth_factor", serde_json::json!(enabled))
.await
.expect("Failed to set email_auth_factor");
}
async fn get_handle(did: &str) -> String {
let repos = get_test_repos().await;
repos
.user
.get_handle_by_did(&Did::new(did.to_string()).unwrap())
.await
.expect("DB error")
.expect("Handle not found")
.to_string()
}
#[tokio::test]
@@ -102,12 +109,7 @@ async fn test_legacy_2fa_auth_factor_required() {
enable_totp_for_user(&did).await;
set_allow_legacy_login(&did, true).await;
let pool = get_test_db_pool().await;
let handle: String = sqlx::query_scalar::<_, String>("SELECT handle FROM users WHERE did = $1")
.bind(&did)
.fetch_one(pool)
.await
.expect("Failed to get handle");
let handle = get_handle(&did).await;
let login_payload = json!({
"identifier": handle,
@@ -141,12 +143,7 @@ async fn test_legacy_2fa_valid_code_succeeds() {
set_allow_legacy_login(&did, true).await;
clear_2fa_challenges_for_user(&did).await;
let pool = get_test_db_pool().await;
let handle: String = sqlx::query_scalar::<_, String>("SELECT handle FROM users WHERE did = $1")
.bind(&did)
.fetch_one(pool)
.await
.expect("Failed to get handle");
let handle = get_handle(&did).await;
let login_payload = json!({
"identifier": handle,
@@ -194,12 +191,7 @@ async fn test_legacy_2fa_invalid_code_rejected() {
set_allow_legacy_login(&did, true).await;
clear_2fa_challenges_for_user(&did).await;
let pool = get_test_db_pool().await;
let handle: String = sqlx::query_scalar::<_, String>("SELECT handle FROM users WHERE did = $1")
.bind(&did)
.fetch_one(pool)
.await
.expect("Failed to get handle");
let handle = get_handle(&did).await;
let resp = client
.post(format!("{}/xrpc/com.atproto.server.createSession", base))
@@ -245,12 +237,7 @@ async fn test_legacy_2fa_blocked_when_disabled() {
enable_totp_for_user(&did).await;
set_allow_legacy_login(&did, false).await;
let pool = get_test_db_pool().await;
let handle: String = sqlx::query_scalar::<_, String>("SELECT handle FROM users WHERE did = $1")
.bind(&did)
.fetch_one(pool)
.await
.expect("Failed to get handle");
let handle = get_handle(&did).await;
let login_payload = json!({
"identifier": handle,
@@ -274,12 +261,7 @@ async fn test_legacy_2fa_no_totp_no_challenge() {
let base = base_url().await;
let (_token, did) = create_account_and_login(&client).await;
let pool = get_test_db_pool().await;
let handle: String = sqlx::query_scalar::<_, String>("SELECT handle FROM users WHERE did = $1")
.bind(&did)
.fetch_one(pool)
.await
.expect("Failed to get handle");
let handle = get_handle(&did).await;
let login_payload = json!({
"identifier": handle,
@@ -307,12 +289,7 @@ async fn test_legacy_2fa_code_consumed_after_use() {
set_allow_legacy_login(&did, true).await;
clear_2fa_challenges_for_user(&did).await;
let pool = get_test_db_pool().await;
let handle: String = sqlx::query_scalar::<_, String>("SELECT handle FROM users WHERE did = $1")
.bind(&did)
.fetch_one(pool)
.await
.expect("Failed to get handle");
let handle = get_handle(&did).await;
let resp = client
.post(format!("{}/xrpc/com.atproto.server.createSession", base))
@@ -404,12 +381,7 @@ async fn test_email_auth_factor_requires_code() {
set_email_auth_factor(&did, true).await;
clear_2fa_challenges_for_user(&did).await;
let pool = get_test_db_pool().await;
let handle: String = sqlx::query_scalar::<_, String>("SELECT handle FROM users WHERE did = $1")
.bind(&did)
.fetch_one(pool)
.await
.expect("Failed to get handle");
let handle = get_handle(&did).await;
let login_payload = json!({
"identifier": handle,
@@ -457,12 +429,7 @@ async fn test_email_auth_factor_disabled_no_challenge() {
set_email_auth_factor(&did, false).await;
let pool = get_test_db_pool().await;
let handle: String = sqlx::query_scalar::<_, String>("SELECT handle FROM users WHERE did = $1")
.bind(&did)
.fetch_one(pool)
.await
.expect("Failed to get handle");
let handle = get_handle(&did).await;
let login_payload = json!({
"identifier": handle,
+18 -14
View File
@@ -577,19 +577,23 @@ async fn test_request_account_delete() {
.await
.expect("Failed to request account deletion");
assert_eq!(res.status(), StatusCode::OK);
let db_url = get_db_connection_string().await;
let pool = sqlx::PgPool::connect(&db_url)
let repos = get_test_repos().await;
let deletion_request = repos
.infra
.get_deletion_request_by_did(&tranquil_types::Did::new(did.clone()).unwrap())
.await
.expect("Failed to connect to test DB");
let row = sqlx::query!(
"SELECT token, expires_at FROM account_deletion_requests WHERE did = $1",
did
)
.fetch_optional(&pool)
.await
.expect("Failed to query DB");
assert!(row.is_some(), "Deletion token should exist in DB");
let row = row.unwrap();
assert!(!row.token.is_empty(), "Token should not be empty");
assert!(row.expires_at > Utc::now(), "Token should not be expired");
.expect("Failed to query DB");
assert!(
deletion_request.is_some(),
"Deletion token should exist in DB"
);
let deletion_request = deletion_request.unwrap();
assert!(
!deletion_request.token.is_empty(),
"Token should not be empty"
);
assert!(
deletion_request.expires_at > Utc::now(),
"Token should not be expired"
);
}
+63 -74
View File
@@ -1,91 +1,80 @@
mod common;
use sqlx::Row;
use tranquil_pds::comms::{CommsChannel, CommsStatus, CommsType};
use tranquil_db_traits::{CommsChannel, CommsStatus, CommsType};
use tranquil_types::Did;
#[tokio::test]
async fn test_enqueue_comms() {
let pool = common::get_test_db_pool().await;
let repos = common::get_test_repos().await;
let (_, did) = common::create_account_and_login(&common::client()).await;
let user_id: uuid::Uuid = sqlx::query_scalar("SELECT id FROM users WHERE did = $1")
.bind(&did)
.fetch_one(pool)
let user_id = repos
.user
.get_id_by_did(&Did::new(did).unwrap())
.await
.expect("DB error")
.expect("User not found");
let comms_id: uuid::Uuid = sqlx::query_scalar(
r#"INSERT INTO comms_queue (user_id, channel, comms_type, recipient, subject, body)
VALUES ($1, 'email', 'welcome', $2, $3, $4)
RETURNING id"#,
)
.bind(user_id)
.bind("test@example.com")
.bind("Test Subject")
.bind("Test body")
.fetch_one(pool)
.await
.expect("Failed to enqueue comms");
let row = sqlx::query(
r#"
SELECT id, user_id, recipient, subject, body, channel, comms_type, status
FROM comms_queue
WHERE id = $1
"#,
)
.bind(comms_id)
.fetch_one(pool)
.await
.expect("Comms not found");
let row_user_id: uuid::Uuid = row.get("user_id");
let row_recipient: String = row.get("recipient");
let row_subject: Option<String> = row.get("subject");
let row_body: String = row.get("body");
let row_channel: CommsChannel = row.get("channel");
let row_comms_type: CommsType = row.get("comms_type");
let row_status: CommsStatus = row.get("status");
assert_eq!(row_user_id, user_id);
assert_eq!(row_recipient, "test@example.com");
assert_eq!(row_subject.as_deref(), Some("Test Subject"));
assert_eq!(row_body, "Test body");
assert_eq!(row_channel, CommsChannel::Email);
assert_eq!(row_comms_type, CommsType::Welcome);
assert_eq!(row_status, CommsStatus::Pending);
repos
.infra
.enqueue_comms(
Some(user_id),
CommsChannel::Email,
CommsType::Welcome,
"test@example.com",
Some("Test Subject"),
"Test body",
None,
)
.await
.expect("Failed to enqueue comms");
let comms = repos
.infra
.get_latest_comms_for_user(user_id, CommsType::Welcome, 1)
.await
.expect("DB error");
let row = comms.first().expect("Comms not found");
assert_eq!(row.user_id, Some(user_id));
assert_eq!(row.recipient, "test@example.com");
assert_eq!(row.subject.as_deref(), Some("Test Subject"));
assert_eq!(row.body, "Test body");
assert_eq!(row.channel, CommsChannel::Email);
assert_eq!(row.comms_type, CommsType::Welcome);
assert_eq!(row.status, CommsStatus::Pending);
}
#[tokio::test]
async fn test_comms_queue_status_index() {
let pool = common::get_test_db_pool().await;
let repos = common::get_test_repos().await;
let (_, did) = common::create_account_and_login(&common::client()).await;
let user_id: uuid::Uuid = sqlx::query_scalar("SELECT id FROM users WHERE did = $1")
.bind(&did)
.fetch_one(pool)
let user_id = repos
.user
.get_id_by_did(&Did::new(did).unwrap())
.await
.expect("DB error")
.expect("User not found");
let initial_count: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM comms_queue WHERE status = 'pending' AND user_id = $1",
)
.bind(user_id)
.fetch_one(pool)
.await
.expect("Failed to count");
let inserts = (0..5).map(|i| {
sqlx::query(
r#"INSERT INTO comms_queue (user_id, channel, comms_type, recipient, subject, body)
VALUES ($1, 'email', 'password_reset', $2, $3, $4)"#,
)
.bind(user_id)
.bind(format!("test{}@example.com", i))
.bind("Test")
.bind("Body")
.execute(pool)
});
futures::future::try_join_all(inserts)
let initial_count = repos
.infra
.count_comms_by_type(user_id, CommsType::PasswordReset)
.await
.expect("Failed to enqueue");
let final_count: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM comms_queue WHERE status = 'pending' AND user_id = $1",
)
.bind(user_id)
.fetch_one(pool)
.await
.expect("Failed to count");
.expect("Failed to count");
for i in 0..5 {
let recipient = format!("test{}@example.com", i);
repos
.infra
.enqueue_comms(
Some(user_id),
CommsChannel::Email,
CommsType::PasswordReset,
&recipient,
Some("Test"),
"Body",
None,
)
.await
.expect("Failed to enqueue");
}
let final_count = repos
.infra
.count_comms_by_type(user_id, CommsType::PasswordReset)
.await
.expect("Failed to count");
assert_eq!(final_count - initial_count, 5);
}
+26 -25
View File
@@ -1,11 +1,12 @@
mod common;
mod helpers;
use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
use common::{base_url, client, get_test_db_pool};
use common::{base_url, client, get_test_repos};
use helpers::verify_new_account;
use reqwest::{StatusCode, redirect};
use serde_json::{Value, json};
use sha2::{Digest, Sha256};
use tranquil_types::{Did, RequestId};
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
@@ -449,10 +450,10 @@ async fn test_oauth_2fa_flow() {
let account: Value = create_res.json().await.unwrap();
let user_did = account["did"].as_str().unwrap();
verify_new_account(&http_client, user_did).await;
let pool = get_test_db_pool().await;
sqlx::query("UPDATE users SET two_factor_enabled = true WHERE did = $1")
.bind(user_did)
.execute(pool)
let repos = get_test_repos().await;
repos
.user
.set_two_factor_enabled(&Did::new(user_did.to_string()).unwrap(), true)
.await
.unwrap();
let redirect_uri = "https://example.com/2fa-callback";
@@ -508,12 +509,12 @@ async fn test_oauth_2fa_flow() {
.contains("Invalid")
|| body["error"].as_str().unwrap_or("") == "invalid_code"
);
let twofa_code: String =
sqlx::query_scalar("SELECT code FROM oauth_2fa_challenge WHERE request_uri = $1")
.bind(request_uri)
.fetch_one(pool)
.await
.unwrap();
let twofa_code: String = repos
.oauth
.get_2fa_challenge_code(&RequestId::new(request_uri.to_string()))
.await
.unwrap()
.unwrap();
let twofa_res = http_client
.post(format!("{}/oauth/authorize/2fa", url))
.header("Content-Type", "application/json")
@@ -574,10 +575,10 @@ async fn test_oauth_2fa_lockout() {
let account: Value = create_res.json().await.unwrap();
let user_did = account["did"].as_str().unwrap();
verify_new_account(&http_client, user_did).await;
let pool = get_test_db_pool().await;
sqlx::query("UPDATE users SET two_factor_enabled = true WHERE did = $1")
.bind(user_did)
.execute(pool)
let repos = get_test_repos().await;
repos
.user
.set_two_factor_enabled(&Did::new(user_did.to_string()).unwrap(), true)
.await
.unwrap();
let redirect_uri = "https://example.com/2fa-lockout-callback";
@@ -748,10 +749,10 @@ async fn test_account_selector_with_2fa() {
.json::<Value>()
.await
.unwrap();
let pool = get_test_db_pool().await;
sqlx::query("UPDATE users SET two_factor_enabled = true WHERE did = $1")
.bind(&user_did)
.execute(pool)
let repos = get_test_repos().await;
repos
.user
.set_two_factor_enabled(&Did::new(user_did.to_string()).unwrap(), true)
.await
.unwrap();
let (code_verifier2, code_challenge2) = generate_pkce();
@@ -789,12 +790,12 @@ async fn test_account_selector_with_2fa() {
select_body["needs_2fa"].as_bool().unwrap_or(false),
"Should need 2FA"
);
let twofa_code: String =
sqlx::query_scalar("SELECT code FROM oauth_2fa_challenge WHERE request_uri = $1")
.bind(request_uri2)
.fetch_one(pool)
.await
.unwrap();
let twofa_code: String = repos
.oauth
.get_2fa_challenge_code(&RequestId::new(request_uri2.to_string()))
.await
.unwrap()
.unwrap();
let twofa_res = http_client
.post(format!("{}/oauth/authorize/2fa", url))
.header("cookie", &device_cookie)
+64 -74
View File
@@ -3,12 +3,13 @@ mod helpers;
use helpers::verify_new_account;
use reqwest::StatusCode;
use serde_json::{Value, json};
use tranquil_db_traits::CommsType;
#[tokio::test]
async fn test_request_password_reset_creates_code() {
let client = common::client();
let base_url = common::base_url().await;
let pool = common::get_test_db_pool().await;
let repos = common::get_test_repos().await;
let handle = format!("pr{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
let email = format!("{}@example.com", handle);
let payload = json!({
@@ -36,16 +37,15 @@ async fn test_request_password_reset_creates_code() {
.await
.expect("Failed to request password reset");
assert_eq!(res.status(), StatusCode::OK);
let user = sqlx::query!(
"SELECT password_reset_code, password_reset_code_expires_at FROM users WHERE email = $1",
email
)
.fetch_one(pool)
.await
.expect("User not found");
assert!(user.password_reset_code.is_some());
assert!(user.password_reset_code_expires_at.is_some());
let code = user.password_reset_code.unwrap();
let info = repos
.user
.get_password_reset_info(&email)
.await
.expect("failed to look up user")
.expect("user not found");
assert!(info.code.is_some());
assert!(info.expires_at.is_some());
let code = info.code.unwrap();
assert!(code.contains('-'));
assert_eq!(code.len(), 11);
}
@@ -70,7 +70,7 @@ async fn test_request_password_reset_unknown_email_returns_ok() {
async fn test_reset_password_with_valid_token() {
let client = common::client();
let base_url = common::base_url().await;
let pool = common::get_test_db_pool().await;
let repos = common::get_test_repos().await;
let handle = format!("pr2{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
let email = format!("{}@example.com", handle);
let old_password = "Oldpass123!";
@@ -103,14 +103,13 @@ async fn test_reset_password_with_valid_token() {
.await
.expect("Failed to request password reset");
assert_eq!(res.status(), StatusCode::OK);
let user = sqlx::query!(
"SELECT password_reset_code FROM users WHERE email = $1",
email
)
.fetch_one(pool)
.await
.expect("User not found");
let token = user.password_reset_code.expect("No reset code");
let info = repos
.user
.get_password_reset_info(&email)
.await
.expect("failed to look up user")
.expect("user not found");
let token = info.code.expect("No reset code");
let res = client
.post(format!(
"{}/xrpc/com.atproto.server.resetPassword",
@@ -124,15 +123,14 @@ async fn test_reset_password_with_valid_token() {
.await
.expect("Failed to reset password");
assert_eq!(res.status(), StatusCode::OK);
let user = sqlx::query!(
"SELECT password_reset_code, password_reset_code_expires_at FROM users WHERE email = $1",
email
)
.fetch_one(pool)
.await
.expect("User not found");
assert!(user.password_reset_code.is_none());
assert!(user.password_reset_code_expires_at.is_none());
let info = repos
.user
.get_password_reset_info(&email)
.await
.expect("failed to look up user")
.expect("user not found");
assert!(info.code.is_none());
assert!(info.expires_at.is_none());
let res = client
.post(format!(
"{}/xrpc/com.atproto.server.createSession",
@@ -186,7 +184,7 @@ async fn test_reset_password_with_invalid_token() {
async fn test_reset_password_with_expired_token() {
let client = common::client();
let base_url = common::base_url().await;
let pool = common::get_test_db_pool().await;
let repos = common::get_test_repos().await;
let handle = format!("pr3{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
let email = format!("{}@example.com", handle);
let payload = json!({
@@ -214,21 +212,18 @@ async fn test_reset_password_with_expired_token() {
.await
.expect("Failed to request password reset");
assert_eq!(res.status(), StatusCode::OK);
let user = sqlx::query!(
"SELECT password_reset_code FROM users WHERE email = $1",
email
)
.fetch_one(pool)
.await
.expect("User not found");
let token = user.password_reset_code.expect("No reset code");
sqlx::query!(
"UPDATE users SET password_reset_code_expires_at = NOW() - INTERVAL '1 hour' WHERE email = $1",
email
)
.execute(pool)
.await
.expect("Failed to expire token");
let info = repos
.user
.get_password_reset_info(&email)
.await
.expect("failed to look up user")
.expect("user not found");
let token = info.code.expect("No reset code");
repos
.user
.expire_password_reset_code(&email)
.await
.expect("Failed to expire token");
let res = client
.post(format!(
"{}/xrpc/com.atproto.server.resetPassword",
@@ -250,7 +245,7 @@ async fn test_reset_password_with_expired_token() {
async fn test_reset_password_invalidates_sessions() {
let client = common::client();
let base_url = common::base_url().await;
let pool = common::get_test_db_pool().await;
let repos = common::get_test_repos().await;
let handle = format!("pr4{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
let email = format!("{}@example.com", handle);
let payload = json!({
@@ -288,14 +283,13 @@ async fn test_reset_password_invalidates_sessions() {
.await
.expect("Failed to request password reset");
assert_eq!(res.status(), StatusCode::OK);
let user = sqlx::query!(
"SELECT password_reset_code FROM users WHERE email = $1",
email
)
.fetch_one(pool)
.await
.expect("User not found");
let token = user.password_reset_code.expect("No reset code");
let info = repos
.user
.get_password_reset_info(&email)
.await
.expect("failed to look up user")
.expect("user not found");
let token = info.code.expect("No reset code");
let res = client
.post(format!(
"{}/xrpc/com.atproto.server.resetPassword",
@@ -338,7 +332,7 @@ async fn test_request_password_reset_empty_email() {
#[tokio::test]
async fn test_reset_password_creates_notification() {
let pool = common::get_test_db_pool().await;
let repos = common::get_test_repos().await;
let client = common::client();
let base_url = common::base_url().await;
let handle = format!("pr5{}", &uuid::Uuid::new_v4().simple().to_string()[..12]);
@@ -358,18 +352,17 @@ async fn test_reset_password_creates_notification() {
.await
.expect("Failed to create account");
assert_eq!(res.status(), StatusCode::OK);
let user = sqlx::query!("SELECT id FROM users WHERE email = $1", email)
.fetch_one(pool)
let user = repos
.user
.get_by_email(&email)
.await
.expect("User not found");
let initial_count: i64 = sqlx::query_scalar!(
"SELECT COUNT(*) FROM comms_queue WHERE user_id = $1 AND comms_type = 'password_reset'",
user.id
)
.fetch_one(pool)
.await
.expect("Failed to count")
.unwrap_or(0);
.expect("failed to look up user")
.expect("user not found");
let initial_count = repos
.infra
.count_comms_by_type(user.id, CommsType::PasswordReset)
.await
.expect("Failed to count");
let res = client
.post(format!(
"{}/xrpc/com.atproto.server.requestPasswordReset",
@@ -380,13 +373,10 @@ async fn test_reset_password_creates_notification() {
.await
.expect("Failed to request password reset");
assert_eq!(res.status(), StatusCode::OK);
let final_count: i64 = sqlx::query_scalar!(
"SELECT COUNT(*) FROM comms_queue WHERE user_id = $1 AND comms_type = 'password_reset'",
user.id
)
.fetch_one(pool)
.await
.expect("Failed to count")
.unwrap_or(0);
let final_count = repos
.infra
.count_comms_by_type(user.id, CommsType::PasswordReset)
.await
.expect("Failed to count");
assert_eq!(final_count - initial_count, 1);
}
+15 -34
View File
@@ -3,7 +3,7 @@ use common::*;
use k256::ecdsa::SigningKey;
use reqwest::StatusCode;
use serde_json::{Value, json};
use sqlx::PgPool;
use tranquil_types::Did;
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
@@ -36,47 +36,28 @@ fn get_multikey_from_signing_key(signing_key: &SigningKey) -> String {
}
async fn get_user_signing_key(did: &str) -> Option<Vec<u8>> {
let db_url = get_db_connection_string().await;
let pool = PgPool::connect(&db_url).await.ok()?;
let row = sqlx::query!(
r#"
SELECT k.key_bytes, k.encryption_version
FROM user_keys k
JOIN users u ON k.user_id = u.id
WHERE u.did = $1
"#,
did
)
.fetch_optional(&pool)
.await
.ok()??;
tranquil_pds::config::decrypt_key(&row.key_bytes, row.encryption_version).ok()
let repos = get_test_repos().await;
let parsed_did = Did::new(did.to_string()).ok()?;
let key_info = repos.user.get_user_key_by_did(&parsed_did).await.ok()??;
tranquil_pds::config::decrypt_key(&key_info.key_bytes, key_info.encryption_version).ok()
}
async fn get_plc_token_from_db(did: &str) -> Option<String> {
let db_url = get_db_connection_string().await;
let pool = PgPool::connect(&db_url).await.ok()?;
sqlx::query_scalar!(
r#"
SELECT t.token
FROM plc_operation_tokens t
JOIN users u ON t.user_id = u.id
WHERE u.did = $1
"#,
did
)
.fetch_optional(&pool)
.await
.ok()?
let repos = get_test_repos().await;
let parsed_did = Did::new(did.to_string()).ok()?;
let tokens = repos.infra.get_plc_tokens_by_did(&parsed_did).await.ok()?;
tokens.into_iter().next().map(|t| t.token)
}
async fn get_user_handle(did: &str) -> Option<String> {
let db_url = get_db_connection_string().await;
let pool = PgPool::connect(&db_url).await.ok()?;
sqlx::query_scalar!(r#"SELECT handle FROM users WHERE did = $1"#, did)
.fetch_optional(&pool)
let repos = get_test_repos().await;
let parsed_did = Did::new(did.to_string()).ok()?;
repos
.user
.get_handle_by_did(&parsed_did)
.await
.ok()?
.map(|h| h.to_string())
}
fn create_mock_last_op(
+37 -21
View File
@@ -2,7 +2,7 @@ mod common;
use common::*;
use reqwest::StatusCode;
use serde_json::json;
use sqlx::PgPool;
use tranquil_types::Did;
#[tokio::test]
async fn test_plc_operation_auth() {
@@ -176,26 +176,34 @@ async fn test_plc_token_lifecycle() {
.await
.unwrap();
assert_eq!(res.status(), StatusCode::OK);
let db_url = get_db_connection_string().await;
let pool = PgPool::connect(&db_url).await.unwrap();
let row = sqlx::query!(
"SELECT t.token, t.expires_at FROM plc_operation_tokens t JOIN users u ON t.user_id = u.id WHERE u.did = $1",
did
).fetch_optional(&pool).await.unwrap();
assert!(row.is_some(), "PLC token should be created in database");
let row = row.unwrap();
assert_eq!(row.token.len(), 11, "Token should be in format xxxxx-xxxxx");
assert!(row.token.contains('-'), "Token should contain hyphen");
let repos = get_test_repos().await;
let parsed_did = Did::new(did.clone()).unwrap();
let tokens = repos
.infra
.get_plc_tokens_by_did(&parsed_did)
.await
.unwrap();
assert!(
row.expires_at > chrono::Utc::now(),
!tokens.is_empty(),
"PLC token should be created in database"
);
let first = &tokens[0];
assert_eq!(
first.token.len(),
11,
"Token should be in format xxxxx-xxxxx"
);
assert!(first.token.contains('-'), "Token should contain hyphen");
assert!(
first.expires_at > chrono::Utc::now(),
"Token should not be expired"
);
let diff = row.expires_at - chrono::Utc::now();
let diff = first.expires_at - chrono::Utc::now();
assert!(
diff.num_minutes() >= 9 && diff.num_minutes() <= 11,
"Token should expire in ~10 minutes"
);
let token1 = row.token.clone();
let token1 = first.token.clone();
let res = client
.post(format!(
"{}/xrpc/com.atproto.identity.requestPlcOperationSignature",
@@ -206,12 +214,20 @@ async fn test_plc_token_lifecycle() {
.await
.unwrap();
assert_eq!(res.status(), StatusCode::OK);
let token2 = sqlx::query_scalar!(
"SELECT t.token FROM plc_operation_tokens t JOIN users u ON t.user_id = u.id WHERE u.did = $1", did
).fetch_one(&pool).await.unwrap();
assert_ne!(token1, token2, "Second request should generate a new token");
let count: i64 = sqlx::query_scalar!(
"SELECT COUNT(*) as \"count!\" FROM plc_operation_tokens t JOIN users u ON t.user_id = u.id WHERE u.did = $1", did
).fetch_one(&pool).await.unwrap();
let tokens2 = repos
.infra
.get_plc_tokens_by_did(&parsed_did)
.await
.unwrap();
let token2 = &tokens2[0].token;
assert_ne!(
token1, *token2,
"Second request should generate a new token"
);
let count = repos
.infra
.count_plc_tokens_by_did(&parsed_did)
.await
.unwrap();
assert_eq!(count, 1, "Should only have one token per user");
}
+16 -41
View File
@@ -58,11 +58,8 @@ async fn test_create_record_cid_matches_firehose() {
let client = client();
let (token, did) = create_account_and_login(&client).await;
let pool = get_test_db_pool().await;
let cursor: i64 = sqlx::query_scalar::<_, i64>("SELECT COALESCE(MAX(seq), 0) FROM repo_seq")
.fetch_one(pool)
.await
.unwrap();
let repos = get_test_repos().await;
let cursor = repos.repo.get_max_seq().await.unwrap().as_i64();
let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await;
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
@@ -136,11 +133,8 @@ async fn test_update_record_prev_matches_old_cid() {
let v1_cid_str = v1_body["cid"].as_str().unwrap();
let v1_cid = Cid::from_str(v1_cid_str).unwrap();
let pool = get_test_db_pool().await;
let cursor: i64 = sqlx::query_scalar::<_, i64>("SELECT COALESCE(MAX(seq), 0) FROM repo_seq")
.fetch_one(pool)
.await
.unwrap();
let repos = get_test_repos().await;
let cursor = repos.repo.get_max_seq().await.unwrap().as_i64();
let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await;
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
@@ -208,11 +202,8 @@ async fn test_delete_record_prev_set_cid_none() {
let collection = parts[parts.len() - 2];
let rkey = parts[parts.len() - 1];
let pool = get_test_db_pool().await;
let cursor: i64 = sqlx::query_scalar::<_, i64>("SELECT COALESCE(MAX(seq), 0) FROM repo_seq")
.fetch_one(pool)
.await
.unwrap();
let repos = get_test_repos().await;
let cursor = repos.repo.get_max_seq().await.unwrap().as_i64();
let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await;
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
@@ -254,11 +245,8 @@ async fn test_five_record_commit_chain_integrity() {
let client = client();
let (token, did) = create_account_and_login(&client).await;
let pool = get_test_db_pool().await;
let cursor: i64 = sqlx::query_scalar::<_, i64>("SELECT COALESCE(MAX(seq), 0) FROM repo_seq")
.fetch_one(pool)
.await
.unwrap();
let repos = get_test_repos().await;
let cursor = repos.repo.get_max_seq().await.unwrap().as_i64();
let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await;
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
@@ -326,11 +314,8 @@ async fn test_apply_writes_single_commit_multiple_ops() {
let client = client();
let (token, did) = create_account_and_login(&client).await;
let pool = get_test_db_pool().await;
let cursor: i64 = sqlx::query_scalar::<_, i64>("SELECT COALESCE(MAX(seq), 0) FROM repo_seq")
.fetch_one(pool)
.await
.unwrap();
let repos = get_test_repos().await;
let cursor = repos.repo.get_max_seq().await.unwrap().as_i64();
let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await;
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
@@ -410,11 +395,8 @@ async fn test_firehose_commit_signature_verification() {
bytes: std::borrow::Cow::Owned(pubkey_bytes.as_bytes().to_vec()),
};
let pool = get_test_db_pool().await;
let cursor: i64 = sqlx::query_scalar::<_, i64>("SELECT COALESCE(MAX(seq), 0) FROM repo_seq")
.fetch_one(pool)
.await
.unwrap();
let repos = get_test_repos().await;
let cursor = repos.repo.get_max_seq().await.unwrap().as_i64();
let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await;
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
@@ -461,12 +443,8 @@ async fn test_cursor_backfill_completeness() {
let client = client();
let (token, did) = create_account_and_login(&client).await;
let pool = get_test_db_pool().await;
let baseline_seq: i64 =
sqlx::query_scalar::<_, i64>("SELECT COALESCE(MAX(seq), 0) FROM repo_seq")
.fetch_one(pool)
.await
.unwrap();
let repos = get_test_repos().await;
let baseline_seq = repos.repo.get_max_seq().await.unwrap().as_i64();
let mut expected_cids: Vec<String> = Vec::with_capacity(5);
let texts = [
@@ -517,11 +495,8 @@ async fn test_multi_account_seq_interleaving() {
let (alice_token, alice_did) = create_account_and_login(&client).await;
let (bob_token, bob_did) = create_account_and_login(&client).await;
let pool = get_test_db_pool().await;
let cursor: i64 = sqlx::query_scalar::<_, i64>("SELECT COALESCE(MAX(seq), 0) FROM repo_seq")
.fetch_one(pool)
.await
.unwrap();
let repos = get_test_repos().await;
let cursor = repos.repo.get_max_seq().await.unwrap().as_i64();
let consumer = FirehoseConsumer::connect_with_cursor(app_port(), cursor).await;
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+32 -16
View File
@@ -97,14 +97,22 @@ async fn cluster_any_node_access() {
.expect("no accessJwt")
.to_string();
let pool = common::get_test_db_pool().await;
let body_text: String = sqlx::query_scalar!(
"SELECT body FROM comms_queue WHERE user_id = (SELECT id FROM users WHERE did = $1) AND comms_type = 'email_verification' ORDER BY created_at DESC LIMIT 1",
&did
)
.fetch_one(pool)
.await
.expect("verification code not found");
let repos = common::get_test_repos().await;
let user = repos
.user
.get_by_did(&tranquil_types::Did::new(did.clone()).unwrap())
.await
.expect("failed to look up user")
.expect("user not found");
let comms = repos
.infra
.get_latest_comms_for_user(user.id, tranquil_db_traits::CommsType::EmailVerification, 1)
.await
.expect("failed to get comms");
let body_text = comms
.first()
.map(|c| c.body.clone())
.expect("no email_verification comms found");
let lines: Vec<&str> = body_text.lines().collect();
let verification_code = lines
@@ -624,14 +632,22 @@ fn create_account_on_node<'a>(
.expect("no accessJwt")
.to_string();
let pool = common::get_test_db_pool().await;
let body_text: String = sqlx::query_scalar!(
"SELECT body FROM comms_queue WHERE user_id = (SELECT id FROM users WHERE did = $1) AND comms_type = 'email_verification' ORDER BY created_at DESC LIMIT 1",
&did
)
.fetch_one(pool)
.await
.expect("verification code not found");
let repos = common::get_test_repos().await;
let user = repos
.user
.get_by_did(&tranquil_types::Did::new(did.clone()).unwrap())
.await
.expect("failed to look up user")
.expect("user not found");
let comms = repos
.infra
.get_latest_comms_for_user(user.id, tranquil_db_traits::CommsType::EmailVerification, 1)
.await
.expect("failed to get comms");
let body_text = comms
.first()
.map(|c| c.body.clone())
.expect("no email_verification comms found");
let lines: Vec<&str> = body_text.lines().collect();
let verification_code = lines
+22 -25
View File
@@ -31,7 +31,7 @@ async fn test_reserve_signing_key_without_did() {
async fn test_reserve_signing_key_with_did() {
let client = common::client();
let base_url = common::base_url().await;
let pool = common::get_test_db_pool().await;
let repos = common::get_test_repos().await;
let target_did = "did:plc:test123456";
let res = client
.post(format!(
@@ -46,14 +46,13 @@ async fn test_reserve_signing_key_with_did() {
let body: Value = res.json().await.expect("Response was not valid JSON");
let signing_key = body["signingKey"].as_str().unwrap();
assert!(signing_key.starts_with("did:key:z"));
let row = sqlx::query!(
"SELECT did, public_key_did_key FROM reserved_signing_keys WHERE public_key_did_key = $1",
signing_key
)
.fetch_one(pool)
.await
.expect("Reserved key not found in database");
assert_eq!(row.did.as_deref(), Some(target_did));
let row = repos
.infra
.get_reserved_signing_key_full(signing_key)
.await
.expect("db error")
.expect("Reserved key not found in database");
assert_eq!(row.did.as_ref().map(|d| d.as_str()), Some(target_did));
assert_eq!(row.public_key_did_key, signing_key);
}
@@ -61,7 +60,7 @@ async fn test_reserve_signing_key_with_did() {
async fn test_reserve_signing_key_stores_private_key() {
let client = common::client();
let base_url = common::base_url().await;
let pool = common::get_test_db_pool().await;
let repos = common::get_test_repos().await;
let res = client
.post(format!(
"{}/xrpc/com.atproto.server.reserveSigningKey",
@@ -74,13 +73,12 @@ async fn test_reserve_signing_key_stores_private_key() {
assert_eq!(res.status(), StatusCode::OK);
let body: Value = res.json().await.expect("Response was not valid JSON");
let signing_key = body["signingKey"].as_str().unwrap();
let row = sqlx::query!(
"SELECT private_key_bytes, expires_at, used_at FROM reserved_signing_keys WHERE public_key_did_key = $1",
signing_key
)
.fetch_one(pool)
.await
.expect("Reserved key not found in database");
let row = repos
.infra
.get_reserved_signing_key_full(signing_key)
.await
.expect("db error")
.expect("Reserved key not found in database");
assert_eq!(
row.private_key_bytes.len(),
32,
@@ -151,7 +149,7 @@ async fn test_reserve_signing_key_is_public() {
async fn test_create_account_with_reserved_signing_key() {
let client = common::client();
let base_url = common::base_url().await;
let pool = common::get_test_db_pool().await;
let repos = common::get_test_repos().await;
let res = client
.post(format!(
"{}/xrpc/com.atproto.server.reserveSigningKey",
@@ -185,13 +183,12 @@ async fn test_create_account_with_reserved_signing_key() {
let did = body["did"].as_str().unwrap();
let access_jwt = verify_new_account(&client, did).await;
assert!(!access_jwt.is_empty());
let reserved = sqlx::query!(
"SELECT used_at FROM reserved_signing_keys WHERE public_key_did_key = $1",
signing_key
)
.fetch_one(pool)
.await
.expect("Reserved key not found");
let reserved = repos
.infra
.get_reserved_signing_key_full(signing_key)
.await
.expect("db error")
.expect("Reserved key not found");
assert!(
reserved.used_at.is_some(),
"Reserved key should be marked as used"
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
+12 -12
View File
@@ -177,31 +177,31 @@ async fn test_complete_user_journey_signup_to_deletion() {
.expect("Request delete failed");
assert_eq!(request_delete_res.status(), StatusCode::OK);
let pool = get_test_db_pool().await;
let row = sqlx::query!(
"SELECT token FROM account_deletion_requests WHERE did = $1",
did
)
.fetch_one(pool)
.await
.expect("Failed to get deletion token");
let repos = get_test_repos().await;
let deletion_request = repos
.infra
.get_deletion_request_by_did(&tranquil_types::Did::new(did.clone()).unwrap())
.await
.unwrap()
.unwrap();
let final_delete_res = client
.post(format!("{}/xrpc/com.atproto.server.deleteAccount", base))
.json(&json!({
"did": did,
"password": password,
"token": row.token
"token": deletion_request.token
}))
.send()
.await
.expect("Final delete failed");
assert_eq!(final_delete_res.status(), StatusCode::OK);
let user_gone = sqlx::query!("SELECT id FROM users WHERE did = $1", did)
.fetch_optional(pool)
let user_gone = repos
.user
.get_by_did(&tranquil_types::Did::new(did.clone()).unwrap())
.await
.expect("Failed to check user");
.unwrap();
assert!(user_gone.is_none(), "User should be deleted");
}
+2 -2
View File
@@ -114,8 +114,8 @@ async fn run() -> Result<(), Box<dyn std::error::Error>> {
let signal_sender = if tranquil_config::get().signal.enabled {
let slot = Arc::new(tranquil_signal::SignalSlot::default());
state = state.with_signal_sender(slot.clone());
if let Some(client) =
tranquil_signal::SignalClient::from_pool(&state.repos.pool, shutdown.clone()).await
if let Some(provider) = &state.signal_store_provider
&& let Some(client) = provider.load_signal_client(shutdown.clone()).await
{
slot.set_client(client).await;
info!("Signal device already linked");
+4
View File
@@ -4,10 +4,14 @@ version.workspace = true
edition.workspace = true
license.workspace = true
[features]
fjall-store = ["dep:fjall"]
[dependencies]
presage = { workspace = true }
async-trait = { workspace = true }
chrono = { workspace = true }
fjall = { version = "3", optional = true }
sqlx = { workspace = true }
tracing = { workspace = true }
tokio = { workspace = true }

Some files were not shown because too many files have changed in this diff Show More