Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-governance/src/lexigram/ai/governance/relay_ledger/persistence.py: 46%

1"""SQL persistence for the relay ledger (top-ups and check-ins). 

2 

3Only generic ``execute``/``execute_query`` SQL is used so the store 

4works on any backend. Idempotency is enforced by the schema: top-up 

5settlement is compare-and-set on ``status``, and check-ins are 

6PK-guarded on ``(user_id, day)`` so a second same-day award can never 

7be written. 

8""" 

9 

10from __future__ import annotations 

11 

12from typing import TYPE_CHECKING, Any 

13 

14from lexigram.contracts.ai.relay import ( 

15 RelayCheckinRecord, 

16 RelayTopUpRecord, 

17) 

18 

19if TYPE_CHECKING: 

20 from lexigram.contracts.data import DatabaseProviderProtocol 

21 

22__all__ = ["SqlRelayLedgerStore"] 

23 

24_CREATE_TOPUPS = """ 

25CREATE TABLE IF NOT EXISTS ai_relay_topups ( 

26 reference_id TEXT NOT NULL PRIMARY KEY, 

27 user_id TEXT NOT NULL, 

28 amount TEXT NOT NULL, 

29 status TEXT NOT NULL, 

30 created_at TEXT NOT NULL 

31) 

32""" 

33 

34_CREATE_CHECKINS = """ 

35CREATE TABLE IF NOT EXISTS ai_relay_checkins ( 

36 user_id TEXT NOT NULL, 

37 day TEXT NOT NULL, 

38 award TEXT NOT NULL, 

39 created_at TEXT NOT NULL, 

40 PRIMARY KEY (user_id, day) 

41) 

42""" 

43 

44_INSERT_TOPUP = ( 

45 "INSERT INTO ai_relay_topups " 

46 "(reference_id, user_id, amount, status, created_at) " 

47 "VALUES (?, ?, ?, ?, ?)" 

48) 

49 

50_INSERT_CHECKIN = ( 

51 "INSERT OR IGNORE INTO ai_relay_checkins " 

52 "(user_id, day, award, created_at) VALUES (?, ?, ?, ?)" 

53) 

54 

55_SETTLE_TOPUP = ( 

56 "UPDATE ai_relay_topups SET status = 'completed' " 

57 "WHERE reference_id = ? AND status = ?" 

58) 

59 

60_SELECT_TOPUP = "SELECT status FROM ai_relay_topups WHERE reference_id = ?" 

61 

62_LIST_TOPUPS = ( 

63 "SELECT reference_id, user_id, amount, status, created_at " 

64 "FROM ai_relay_topups WHERE user_id = ? " 

65 "ORDER BY created_at DESC, rowid DESC LIMIT ?" 

66) 

67 

68_LIST_TOPUPS_ALL = ( 

69 "SELECT reference_id, user_id, amount, status, created_at " 

70 "FROM ai_relay_topups " 

71 "ORDER BY created_at DESC, rowid DESC LIMIT ?" 

72) 

73 

74 

75def _row_to_topup(row: dict[str, Any]) -> RelayTopUpRecord: 

76 """Rebuild a top-up record from a stored row.""" 

77 return RelayTopUpRecord( 

78 reference_id=row["reference_id"], 

79 user_id=row["user_id"], 

80 amount=row["amount"], 

81 status=row["status"], 

82 created_at=row["created_at"], 

83 ) 

84 

85 

86class SqlRelayLedgerStore: 

87 """SQL-backed ledger rows for top-ups and daily check-ins. 

88 

89 Creates both ledger tables lazily on first use. Callers are the 

90 :class:`~lexigram.ai.governance.relay_ledger.ledger.RelayLedgerService`, 

91 the only permitted mutation path. 

92 

93 Args: 

94 db: A connected 

95 :class:`~lexigram.contracts.data.DatabaseProviderProtocol` 

96 resolved from the DI container. 

97 """ 

98 

99 def __init__(self, db: DatabaseProviderProtocol) -> None: 

100 self._db = db 

101 self._initialised = False 

102 

103 async def _ensure_tables(self) -> None: 

104 """Create the ledger schema once, on first use.""" 

105 if not self._initialised: 

106 await self._db.execute(_CREATE_TOPUPS) 

107 await self._db.execute(_CREATE_CHECKINS) 

108 self._initialised = True 

109 

110 async def insert_topup(self, record: RelayTopUpRecord) -> None: 

111 """Insert one top-up row.""" 

112 await self._ensure_tables() 

113 await self._db.execute( 

114 _INSERT_TOPUP, 

115 [ 

116 record.reference_id, 

117 record.user_id, 

118 record.amount, 

119 record.status, 

120 record.created_at, 

121 ], 

122 ) 

123 

124 async def settle_topup(self, reference_id: str, expected_status: str) -> bool: 

125 """Flip *reference_id* from *expected_status* to completed. 

126 

127 Returns: 

128 ``True`` when exactly one row flipped, ``False`` otherwise. 

129 """ 

130 await self._ensure_tables() 

131 result = await self._db.execute(_SETTLE_TOPUP, [reference_id, expected_status]) 

132 return result.row_count > 0 

133 

134 async def topup_status(self, reference_id: str) -> str | None: 

135 """Return the current status of *reference_id*, or ``None``.""" 

136 await self._ensure_tables() 

137 result = await self._db.execute_query(_SELECT_TOPUP, [reference_id]) 

138 if not result.rows: 

139 return None 

140 return result.rows[0]["status"] 

141 

142 async def insert_checkin(self, record: RelayCheckinRecord) -> bool: 

143 """Insert one check-in row, PK-guarded on ``(user_id, day)``. 

144 

145 Returns: 

146 ``True`` when the award was written, ``False`` when the 

147 user already checked in that day. 

148 """ 

149 await self._ensure_tables() 

150 result = await self._db.execute( 

151 _INSERT_CHECKIN, 

152 [record.user_id, record.day, record.award, record.created_at], 

153 ) 

154 return result.row_count > 0 

155 

156 async def list_topups( 

157 self, user_id: str | None, limit: int 

158 ) -> list[RelayTopUpRecord]: 

159 """Return top-up records, newest first, capped at *limit*. 

160 

161 Args: 

162 user_id: When set, only this user's records; ``None`` 

163 lists every top-up on the ledger. 

164 limit: Maximum rows returned. 

165 """ 

166 await self._ensure_tables() 

167 if user_id is None: 

168 result = await self._db.execute_query(_LIST_TOPUPS_ALL, [limit]) 

169 else: 

170 result = await self._db.execute_query(_LIST_TOPUPS, [user_id, limit]) 

171 return [_row_to_topup(row) for row in result.rows]