shardline_index/postgres/
mod.rs

1mod index_store;
2mod record_store;
3
4use serde_json::Error as JsonError;
5use shardline_protocol::{HashParseError, RangeError};
6use shardline_storage::ObjectKeyError;
7use sqlx::{Error as SqlxError, PgPool};
8use thiserror::Error;
9
10use crate::{QuarantineCandidateError, RetentionHoldError, WebhookDeliveryError};
11
12/// Postgres-compatible implementation of the asynchronous index-store contract.
13#[derive(Debug, Clone)]
14pub struct PostgresIndexStore {
15    pool: PgPool,
16}
17
18impl PostgresIndexStore {
19    /// Creates a Postgres index-store adapter from an existing pool.
20    #[must_use]
21    pub const fn new(pool: PgPool) -> Self {
22        Self { pool }
23    }
24
25    /// Returns the underlying connection pool.
26    #[must_use]
27    pub const fn pool(&self) -> &PgPool {
28        &self.pool
29    }
30}
31
32/// Opaque Postgres file-record locator.
33#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
34pub struct PostgresRecordLocator {
35    record_key: String,
36    kind: PostgresRecordKind,
37    scope_key: String,
38    file_id: String,
39    content_hash: Option<String>,
40}
41
42impl PostgresRecordLocator {
43    /// Returns the stable record key for this locator.
44    #[must_use]
45    pub fn record_key(&self) -> &str {
46        &self.record_key
47    }
48
49    /// Returns the file identifier associated with this locator.
50    #[must_use]
51    pub fn file_id(&self) -> &str {
52        &self.file_id
53    }
54
55    /// Returns the immutable content hash when this locator points at a version record.
56    #[must_use]
57    pub fn content_hash(&self) -> Option<&str> {
58        self.content_hash.as_deref()
59    }
60}
61
62/// Postgres-compatible implementation of the record-store contract.
63#[derive(Debug, Clone)]
64pub struct PostgresRecordStore {
65    pool: PgPool,
66}
67
68impl PostgresRecordStore {
69    /// Creates a Postgres record-store adapter from an existing pool.
70    #[must_use]
71    pub const fn new(pool: PgPool) -> Self {
72        Self { pool }
73    }
74
75    /// Returns the underlying connection pool.
76    #[must_use]
77    pub const fn pool(&self) -> &PgPool {
78        &self.pool
79    }
80}
81
82#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
83pub(super) enum PostgresRecordKind {
84    Latest,
85    Version,
86}
87
88impl PostgresRecordKind {
89    const fn as_str(self) -> &'static str {
90        match self {
91            Self::Latest => "latest",
92            Self::Version => "version",
93        }
94    }
95
96    fn parse(value: &str) -> Result<Self, PostgresMetadataStoreError> {
97        match value {
98            "latest" => Ok(Self::Latest),
99            "version" => Ok(Self::Version),
100            _other => Err(PostgresMetadataStoreError::InvalidRecordKind),
101        }
102    }
103}
104
105/// Postgres metadata-store failure.
106#[derive(Debug, Error)]
107pub enum PostgresMetadataStoreError {
108    /// Postgres access failed.
109    #[error("postgres metadata store operation failed")]
110    Sqlx(#[source] Box<SqlxError>),
111    /// JSON serialization or deserialization failed.
112    #[error("postgres metadata json operation failed")]
113    Json(#[from] JsonError),
114    /// A stored hash value was invalid.
115    #[error("stored hash value was invalid")]
116    HashParse(#[from] HashParseError),
117    /// A stored object key was invalid.
118    #[error("stored object key was invalid")]
119    ObjectKey(#[from] ObjectKeyError),
120    /// A stored chunk range was invalid.
121    #[error("stored chunk range was invalid")]
122    Range(#[from] RangeError),
123    /// A stored retention hold was invalid.
124    #[error("stored retention hold was invalid")]
125    RetentionHold(#[from] RetentionHoldError),
126    /// A stored quarantine candidate was invalid.
127    #[error("stored quarantine candidate was invalid")]
128    QuarantineCandidate(#[from] QuarantineCandidateError),
129    /// A stored webhook delivery was invalid.
130    #[error("stored webhook delivery was invalid")]
131    WebhookDelivery(#[from] WebhookDeliveryError),
132    /// A stored integer exceeded the supported range.
133    #[error("stored integer exceeded the supported range")]
134    IntegerOutOfRange,
135    /// The requested record locator does not exist.
136    #[error("postgres record locator was not found")]
137    RecordNotFound,
138    /// A stored record kind was invalid.
139    #[error("stored record kind was invalid")]
140    InvalidRecordKind,
141    /// An invalid repository type string was encountered.
142    #[error("invalid repository type: {0}")]
143    InvalidRepoType(String),
144}
145
146impl From<SqlxError> for PostgresMetadataStoreError {
147    fn from(value: SqlxError) -> Self {
148        Self::Sqlx(Box::new(value))
149    }
150}
151
152pub(super) fn u64_to_i64(value: u64) -> Result<i64, PostgresMetadataStoreError> {
153    i64::try_from(value).map_err(|_error| PostgresMetadataStoreError::IntegerOutOfRange)
154}
155
156pub(super) fn i64_to_u64(value: i64) -> Result<u64, PostgresMetadataStoreError> {
157    u64::try_from(value).map_err(|_error| PostgresMetadataStoreError::IntegerOutOfRange)
158}
159
160#[cfg(test)]
161mod tests {
162    use shardline_protocol::{ChunkRange, RepositoryProvider, RepositoryScope, ShardlineHash};
163
164    use super::PostgresRecordKind;
165    use super::index_store::PostgresFileReconstructionRecord;
166    use super::record_store::record_locator;
167    use crate::{
168        FileId, FileReconstruction, FileRecord, ReconstructionTerm, RepositoryRecordScope, XorbId,
169        record_key::record_key as shared_record_key,
170        record_key::{
171            repository_record_scope_key as shared_repository_record_scope_key,
172            repository_scope_key as shared_repository_scope_key,
173        },
174        xet_hash_hex_string,
175    };
176
177    #[test]
178    fn postgres_reconstruction_record_roundtrips_domain_terms() {
179        let hash = ShardlineHash::from_bytes([11; 32]);
180        let range = ChunkRange::new(1, 4);
181        assert!(range.is_ok());
182        let Ok(range) = range else {
183            return;
184        };
185        let reconstruction =
186            FileReconstruction::new(vec![ReconstructionTerm::new(XorbId::new(hash), range, 256)]);
187
188        let record = PostgresFileReconstructionRecord::from_domain(&reconstruction);
189        let restored = record.into_domain();
190
191        assert!(matches!(restored, Ok(ref restored) if restored == &reconstruction));
192    }
193
194    #[test]
195    fn postgres_record_keys_distinguish_scope_file_and_kind_without_parsing() {
196        let first_scope =
197            RepositoryScope::new(RepositoryProvider::GitHub, "team:a", "asset", Some("main"));
198        let second_scope =
199            RepositoryScope::new(RepositoryProvider::GitHub, "team", "a:asset", Some("main"));
200        assert!(first_scope.is_ok());
201        assert!(second_scope.is_ok());
202        let (Ok(first_scope), Ok(second_scope)) = (first_scope, second_scope) else {
203            return;
204        };
205
206        let first_key = shared_repository_scope_key(Some(&first_scope));
207        let second_key = shared_repository_scope_key(Some(&second_scope));
208
209        assert_ne!(first_key, second_key);
210        assert_ne!(
211            shared_record_key(
212                PostgresRecordKind::Latest.as_str(),
213                &first_key,
214                "file",
215                None
216            ),
217            shared_record_key(
218                PostgresRecordKind::Version.as_str(),
219                &first_key,
220                "file",
221                Some("a".repeat(64).as_str())
222            )
223        );
224    }
225
226    #[test]
227    fn postgres_latest_locator_ignores_content_hash_for_stable_head_keys() {
228        let scope = RepositoryScope::new(RepositoryProvider::GitLab, "team", "assets", None);
229        assert!(scope.is_ok());
230        let Ok(scope) = scope else {
231            return;
232        };
233        let first = file_record(scope.clone(), "a");
234        let second = file_record(scope, "b");
235        let first_key = record_locator(PostgresRecordKind::Latest, &first, None);
236        let second_key = record_locator(PostgresRecordKind::Latest, &second, None);
237
238        assert_eq!(first_key, second_key);
239    }
240
241    #[test]
242    fn postgres_repository_scope_key_prefix_matches_all_repository_revisions_only() {
243        let repository = RepositoryRecordScope::new(RepositoryProvider::GitHub, "team", "assets");
244        let revisionless = RepositoryScope::new(RepositoryProvider::GitHub, "team", "assets", None);
245        let revisioned =
246            RepositoryScope::new(RepositoryProvider::GitHub, "team", "assets", Some("main"));
247        let other = RepositoryScope::new(RepositoryProvider::GitHub, "team", "other", Some("main"));
248        assert!(revisionless.is_ok());
249        assert!(revisioned.is_ok());
250        assert!(other.is_ok());
251        let (Ok(revisionless), Ok(revisioned), Ok(other)) = (revisionless, revisioned, other)
252        else {
253            return;
254        };
255
256        let repository_key = shared_repository_record_scope_key(&repository);
257        let revisionless_key = shared_repository_scope_key(Some(&revisionless));
258        let revisioned_key = shared_repository_scope_key(Some(&revisioned));
259        let other_key = shared_repository_scope_key(Some(&other));
260
261        assert_eq!(repository_key, revisionless_key);
262        assert!(revisioned_key.starts_with(&repository_key));
263        assert!(!other_key.starts_with(&repository_key));
264    }
265
266    #[test]
267    fn postgres_lifecycle_migrations_reject_inverted_timelines() {
268        let metadata_migration =
269            include_str!("../../migrations/20260417000000_metadata_store.up.sql");
270        let retention_migration =
271            include_str!("../../migrations/20260417010000_retention_holds.up.sql");
272
273        assert!(metadata_migration.contains(
274            "CHECK (delete_after_unix_seconds >= first_seen_unreachable_at_unix_seconds)"
275        ));
276        assert!(retention_migration.contains("release_after_unix_seconds >= held_at_unix_seconds"));
277    }
278
279    fn file_record(scope: RepositoryScope, content_seed: &str) -> FileRecord {
280        let hash = ShardlineHash::from_bytes([12; 32]);
281        let file_id = xet_hash_hex_string(FileId::new(hash).hash());
282        FileRecord {
283            file_id,
284            content_hash: content_seed.repeat(64),
285            total_bytes: 0,
286            chunk_size: 0,
287            repository_scope: Some(scope),
288            chunks: Vec::new(),
289        }
290    }
291}
shardline_index/postgres/mod.rs

shardline_index/postgres/
mod.rs
