sqlite.rs 6.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201
  1. #![allow(missing_docs)]
  2. #![cfg(feature = "sqlite")]
  3. use kuatia_storage::store::{AccountStore, PostingStore};
  4. use kuatia_storage_sql::SqlStore;
  5. use kuatia_types::*;
  6. use sqlx::{Any, Pool, Row};
  7. async fn new_pool() -> Pool<Any> {
  8. sqlx::any::install_default_drivers();
  9. sqlx::any::AnyPoolOptions::new()
  10. .max_connections(1)
  11. .connect("sqlite::memory:")
  12. .await
  13. .unwrap()
  14. }
  15. async fn new_store() -> SqlStore {
  16. let store = SqlStore::new(new_pool().await);
  17. store.migrate().await.unwrap();
  18. store
  19. }
  20. kuatia_storage::store_tests!(new_store);
  21. /// The point of the schema: no column holds opaque binary. A content-addressed
  22. /// id is stored as lower-case hex text, and a structured payload as readable
  23. /// JSON text, so a row can be audited in a plain SQL client.
  24. #[tokio::test]
  25. async fn columns_store_hex_ids_and_json_text() {
  26. let pool = new_pool().await;
  27. let store = SqlStore::new(pool.clone());
  28. store.migrate().await.unwrap();
  29. let account = Account {
  30. id: AccountId::new(1),
  31. version: 1,
  32. policy: AccountPolicy::NoOverdraft,
  33. flags: AccountFlags::empty(),
  34. book: BookId(0),
  35. metadata: std::collections::BTreeMap::new(),
  36. };
  37. store.create_account(account).await.unwrap();
  38. let tid = EnvelopeId([0xab; 32]);
  39. let posting = Posting::new(
  40. PostingId {
  41. transfer: tid,
  42. index: 0,
  43. },
  44. AccountId::new(1),
  45. AssetId::new(1),
  46. Cent::from(100),
  47. );
  48. store.insert_postings(&[posting]).await.unwrap();
  49. // The 32-byte transfer id is stored as its 64-char lower-case hex form.
  50. let row = sqlx::query("SELECT transfer_id FROM postings")
  51. .fetch_one(&pool)
  52. .await
  53. .unwrap();
  54. let transfer_id: String = row.try_get("transfer_id").unwrap();
  55. assert_eq!(transfer_id, "ab".repeat(32));
  56. // The account payload is readable JSON text, not a blob.
  57. let row = sqlx::query("SELECT metadata FROM accounts")
  58. .fetch_one(&pool)
  59. .await
  60. .unwrap();
  61. let metadata: String = row.try_get("metadata").unwrap();
  62. assert!(
  63. serde_json::from_str::<serde_json::Value>(&metadata).is_ok(),
  64. "metadata should be JSON text, got: {metadata}"
  65. );
  66. }
  67. /// The 002 migration adds the subaccount column and a subaccount account/posting
  68. /// round-trips through the schema, kept distinct from the main account.
  69. #[tokio::test]
  70. async fn subaccount_columns_round_trip() {
  71. let store = new_store().await;
  72. let sub = AccountId::with_sub(1, 7);
  73. store
  74. .create_account(Account::new(sub, AccountPolicy::NoOverdraft))
  75. .await
  76. .unwrap();
  77. // The main account (1, 0) is a separate record.
  78. store
  79. .create_account(Account::new(AccountId::new(1), AccountPolicy::NoOverdraft))
  80. .await
  81. .unwrap();
  82. let got = store.get_account(&sub).await.unwrap();
  83. assert_eq!(got.id, sub);
  84. let posting = Posting::new(
  85. PostingId {
  86. transfer: EnvelopeId([0xcd; 32]),
  87. index: 0,
  88. },
  89. sub,
  90. AssetId::new(1),
  91. Cent::from(500),
  92. );
  93. store.insert_postings(&[posting]).await.unwrap();
  94. // Filtering by the subaccount returns it; the main account holds nothing.
  95. let by_sub = store
  96. .get_postings_by_account(1, Some(7), None, None)
  97. .await
  98. .unwrap();
  99. assert_eq!(by_sub.len(), 1);
  100. assert_eq!(by_sub[0].owner, sub);
  101. let by_main = store
  102. .get_postings_by_account(1, Some(0), None, None)
  103. .await
  104. .unwrap();
  105. assert!(by_main.is_empty());
  106. }
  107. /// A database created under 001 (no subaccount column) upgrades in place: the
  108. /// 002 migration rebuilds the tables and existing rows default to subaccount 0,
  109. /// the main account.
  110. #[tokio::test]
  111. async fn migration_upgrades_existing_rows_to_main_account() {
  112. let pool = new_pool().await;
  113. // Pre-002 schema: the tables 002 rebuilds, in their old (no-subaccount) shape.
  114. sqlx::query(
  115. "CREATE TABLE accounts (id BIGINT NOT NULL, version BIGINT NOT NULL, policy TEXT NOT NULL, flags INTEGER NOT NULL, book BIGINT NOT NULL, user_data TEXT NOT NULL, metadata TEXT NOT NULL, PRIMARY KEY (id, version))",
  116. )
  117. .execute(&pool)
  118. .await
  119. .unwrap();
  120. sqlx::query(
  121. "CREATE TABLE postings (transfer_id TEXT NOT NULL, idx SMALLINT NOT NULL, owner BIGINT NOT NULL, asset INTEGER NOT NULL, value TEXT NOT NULL, status SMALLINT NOT NULL, reservation BIGINT, PRIMARY KEY (transfer_id, idx))",
  122. )
  123. .execute(&pool)
  124. .await
  125. .unwrap();
  126. sqlx::query("CREATE INDEX idx_postings_owner ON postings(owner, asset, status)")
  127. .execute(&pool)
  128. .await
  129. .unwrap();
  130. sqlx::query(
  131. "CREATE TABLE transfer_accounts (transfer_id TEXT NOT NULL, account_id BIGINT NOT NULL, PRIMARY KEY (transfer_id, account_id))",
  132. )
  133. .execute(&pool)
  134. .await
  135. .unwrap();
  136. sqlx::query("CREATE INDEX idx_xfer_acct ON transfer_accounts(account_id)")
  137. .execute(&pool)
  138. .await
  139. .unwrap();
  140. // Legacy 001/002 schema carried a user_data JSON column; the 003 migration
  141. // drops it. Seed it with the value the old UserData type serialized to.
  142. let user_data = r#"{"d128":0,"d64":0,"d32":0}"#.to_string();
  143. let metadata =
  144. serde_json::to_string(&std::collections::BTreeMap::<String, String>::new()).unwrap();
  145. sqlx::query(
  146. "INSERT INTO accounts (id, version, policy, flags, book, user_data, metadata) VALUES (5, 1, '\"NoOverdraft\"', 0, 0, $1, $2)",
  147. )
  148. .bind(user_data)
  149. .bind(metadata)
  150. .execute(&pool)
  151. .await
  152. .unwrap();
  153. // Record 001 as applied so migrate() only runs 002 against this schema.
  154. sqlx::query("CREATE TABLE _migrations (name TEXT PRIMARY KEY)")
  155. .execute(&pool)
  156. .await
  157. .unwrap();
  158. sqlx::query("INSERT INTO _migrations (name) VALUES ('001_init')")
  159. .execute(&pool)
  160. .await
  161. .unwrap();
  162. let store = SqlStore::new(pool);
  163. store.migrate().await.unwrap();
  164. // The pre-existing account is now the main account (subaccount 0).
  165. let got = store.get_account(&AccountId::new(5)).await.unwrap();
  166. assert_eq!(got.id, AccountId::new(5));
  167. assert!(got.id.is_main());
  168. }
  169. /// migrate() is idempotent: running it repeatedly on the same DB is a no-op.
  170. #[tokio::test]
  171. async fn migrate_is_idempotent() {
  172. sqlx::any::install_default_drivers();
  173. let pool = sqlx::any::AnyPoolOptions::new()
  174. .max_connections(1)
  175. .connect("sqlite::memory:")
  176. .await
  177. .unwrap();
  178. let store = SqlStore::new(pool);
  179. store.migrate().await.unwrap();
  180. store.migrate().await.unwrap();
  181. store.migrate().await.unwrap();
  182. }