systemprompt_files/repository/file/
mod.rs

1//! Core [`FileRepository`] type and its primary file-row queries.
2//!
3//! Defines the repository over the `files` table — insert/upsert, lookup by id
4//! or path, listing, soft handling, metadata updates, and path search — and
5//! re-exports the [`InsertFileRequest`] builder and [`FileStats`] aggregate.
6//! AI-image and content-association queries live in sibling modules.
7//!
8//! Copyright (c) systemprompt.io — Business Source License 1.1.
9//! See <https://systemprompt.io> for licensing details.
10
11mod request;
12mod stats;
13
14pub use request::InsertFileRequest;
15pub use stats::FileStats;
16
17use std::sync::Arc;
18
19use chrono::Utc;
20use sqlx::PgPool;
21use systemprompt_database::DbPool;
22use systemprompt_identifiers::{ContextId, FileId, SessionId, TraceId, UserId};
23
24use crate::error::{FilesError, FilesResult};
25use crate::models::{File, FileMetadata};
26
27#[derive(Debug, Clone)]
28pub struct FileRepository {
29    pub(crate) pool: Arc<PgPool>,
30    write_pool: Arc<PgPool>,
31}
32
33impl FileRepository {
34    pub fn new(db: &DbPool) -> FilesResult<Self> {
35        let pool = db.pool_arc()?;
36        let write_pool = db.write_pool_arc()?;
37        Ok(Self { pool, write_pool })
38    }
39
40    pub async fn insert(&self, request: InsertFileRequest) -> FilesResult<FileId> {
41        let id_uuid = uuid::Uuid::parse_str(request.id.as_str()).map_err(|e| {
42            FilesError::Validation(format!(
43                "Invalid UUID for file id {}: {e}",
44                request.id.as_str()
45            ))
46        })?;
47        let now = Utc::now();
48
49        let user_id_str = request.user_id.as_ref().map(UserId::as_str);
50        let session_id_str = request.session_id.as_ref().map(SessionId::as_str);
51        let trace_id_str = request.trace_id.as_ref().map(TraceId::as_str);
52        let context_id_str = request.context_id.as_ref().map(ContextId::as_str);
53
54        sqlx::query_as!(
55            File,
56            r#"
57            INSERT INTO files (id, path, public_url, mime_type, size_bytes, ai_content, metadata, user_id, session_id, trace_id, context_id, created_at, updated_at)
58            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $12)
59            ON CONFLICT (path) DO UPDATE SET
60                public_url = EXCLUDED.public_url,
61                mime_type = EXCLUDED.mime_type,
62                size_bytes = EXCLUDED.size_bytes,
63                ai_content = EXCLUDED.ai_content,
64                metadata = EXCLUDED.metadata,
65                updated_at = EXCLUDED.updated_at
66            RETURNING id, path, public_url, mime_type, size_bytes, ai_content, metadata as "metadata: sqlx::types::Json<FileMetadata>", user_id as "user_id: UserId", session_id as "session_id: SessionId", trace_id as "trace_id: TraceId", context_id as "context_id: ContextId", created_at, updated_at, deleted_at
67            "#,
68            id_uuid,
69            request.path,
70            request.public_url,
71            request.mime_type,
72            request.size_bytes,
73            request.ai_content,
74            sqlx::types::Json(&request.metadata) as _,
75            user_id_str,
76            session_id_str,
77            trace_id_str,
78            context_id_str,
79            now
80        )
81        .fetch_one(&*self.write_pool)
82        .await?;
83
84        Ok(request.id)
85    }
86
87    pub async fn insert_file(&self, file: &File) -> FilesResult<FileId> {
88        let file_id = FileId::new(file.id.to_string());
89
90        let mut request = InsertFileRequest::new(
91            file_id.clone(),
92            file.path.clone(),
93            file.public_url.clone(),
94            file.mime_type.clone(),
95        )
96        .with_ai_content(file.ai_content)
97        .with_metadata(file.metadata.0.clone());
98
99        if let Some(size) = file.size_bytes {
100            request = request.with_size(size);
101        }
102
103        if let Some(ref user_id) = file.user_id {
104            request = request.with_user_id(user_id.clone());
105        }
106
107        if let Some(ref session_id) = file.session_id {
108            request = request.with_session_id(session_id.clone());
109        }
110
111        if let Some(ref trace_id) = file.trace_id {
112            request = request.with_trace_id(trace_id.clone());
113        }
114
115        if let Some(ref context_id) = file.context_id {
116            request = request.with_context_id(context_id.clone());
117        }
118
119        self.insert(request).await
120    }
121
122    pub async fn find_by_id(&self, id: &FileId) -> FilesResult<Option<File>> {
123        let id_uuid = uuid::Uuid::parse_str(id.as_str())
124            .map_err(|e| FilesError::Validation(format!("Invalid UUID for file id: {e}")))?;
125
126        let result = sqlx::query_as!(
127            File,
128            r#"
129            SELECT id, path, public_url, mime_type, size_bytes, ai_content, metadata as "metadata: sqlx::types::Json<FileMetadata>", user_id as "user_id: UserId", session_id as "session_id: SessionId", trace_id as "trace_id: TraceId", context_id as "context_id: ContextId", created_at, updated_at, deleted_at
130            FROM files
131            WHERE id = $1 AND deleted_at IS NULL
132            "#,
133            id_uuid
134        )
135        .fetch_optional(&*self.pool)
136        .await?;
137
138        Ok(result)
139    }
140
141    pub async fn find_by_path(&self, path: &str) -> FilesResult<Option<File>> {
142        let result = sqlx::query_as!(
143            File,
144            r#"
145            SELECT id, path, public_url, mime_type, size_bytes, ai_content, metadata as "metadata: sqlx::types::Json<FileMetadata>", user_id as "user_id: UserId", session_id as "session_id: SessionId", trace_id as "trace_id: TraceId", context_id as "context_id: ContextId", created_at, updated_at, deleted_at
146            FROM files
147            WHERE path = $1 AND deleted_at IS NULL
148            "#,
149            path
150        )
151        .fetch_optional(&*self.pool)
152        .await?;
153
154        Ok(result)
155    }
156
157    pub async fn list_by_user(
158        &self,
159        user_id: &UserId,
160        limit: i64,
161        offset: i64,
162    ) -> FilesResult<Vec<File>> {
163        let user_id_str = user_id.as_str();
164        let result = sqlx::query_as!(
165            File,
166            r#"
167            SELECT id, path, public_url, mime_type, size_bytes, ai_content, metadata as "metadata: sqlx::types::Json<FileMetadata>", user_id as "user_id: UserId", session_id as "session_id: SessionId", trace_id as "trace_id: TraceId", context_id as "context_id: ContextId", created_at, updated_at, deleted_at
168            FROM files
169            WHERE user_id = $1 AND deleted_at IS NULL
170            ORDER BY created_at DESC
171            LIMIT $2 OFFSET $3
172            "#,
173            user_id_str,
174            limit,
175            offset
176        )
177        .fetch_all(&*self.pool)
178        .await?;
179
180        Ok(result)
181    }
182
183    pub async fn list_all(&self, limit: i64, offset: i64) -> FilesResult<Vec<File>> {
184        let result = sqlx::query_as!(
185            File,
186            r#"
187            SELECT id, path, public_url, mime_type, size_bytes, ai_content, metadata as "metadata: sqlx::types::Json<FileMetadata>", user_id as "user_id: UserId", session_id as "session_id: SessionId", trace_id as "trace_id: TraceId", context_id as "context_id: ContextId", created_at, updated_at, deleted_at
188            FROM files
189            WHERE deleted_at IS NULL
190            ORDER BY created_at DESC
191            LIMIT $1 OFFSET $2
192            "#,
193            limit,
194            offset
195        )
196        .fetch_all(&*self.pool)
197        .await?;
198
199        Ok(result)
200    }
201
202    pub async fn delete(&self, id: &FileId) -> FilesResult<()> {
203        let id_uuid = uuid::Uuid::parse_str(id.as_str())
204            .map_err(|e| FilesError::Validation(format!("Invalid UUID for file id: {e}")))?;
205
206        sqlx::query!(
207            r#"
208            DELETE FROM files
209            WHERE id = $1
210            "#,
211            id_uuid
212        )
213        .execute(&*self.write_pool)
214        .await?;
215
216        Ok(())
217    }
218
219    pub async fn update_metadata(&self, id: &FileId, metadata: &FileMetadata) -> FilesResult<()> {
220        let id_uuid = uuid::Uuid::parse_str(id.as_str())
221            .map_err(|e| FilesError::Validation(format!("Invalid UUID for file id: {e}")))?;
222        let now = Utc::now();
223
224        sqlx::query!(
225            r#"
226            UPDATE files
227            SET metadata = $1, updated_at = $2
228            WHERE id = $3
229            "#,
230            sqlx::types::Json(metadata) as _,
231            now,
232            id_uuid
233        )
234        .execute(&*self.write_pool)
235        .await?;
236
237        Ok(())
238    }
239
240    pub async fn search_by_path(&self, query: &str, limit: i64) -> FilesResult<Vec<File>> {
241        let pattern = format!("%{query}%");
242        let result = sqlx::query_as!(
243            File,
244            r#"
245            SELECT id, path, public_url, mime_type, size_bytes, ai_content, metadata as "metadata: sqlx::types::Json<FileMetadata>",
246                   user_id as "user_id: UserId", session_id as "session_id: SessionId",
247                   trace_id as "trace_id: TraceId", context_id as "context_id: ContextId",
248                   created_at, updated_at, deleted_at
249            FROM files
250            WHERE path ILIKE $1 AND deleted_at IS NULL
251            ORDER BY created_at DESC
252            LIMIT $2
253            "#,
254            pattern,
255            limit
256        )
257        .fetch_all(&*self.pool)
258        .await?;
259
260        Ok(result)
261    }
262}
systemprompt_files/repository/file/mod.rs

systemprompt_files/repository/file/
mod.rs
