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::{FilesResult, parse_file_uuid};
25use crate::models::{File, FileMetadata, FileRow};
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) -> Self {
35        let pool = db.pool();
36        let write_pool = db.write_pool();
37        Self { pool, write_pool }
38    }
39
40    pub async fn insert(&self, request: InsertFileRequest) -> FilesResult<FileId> {
41        let id_uuid = parse_file_uuid(&request.id)?;
42        let now = Utc::now();
43
44        let user_id_str = request.user_id.as_ref().map(UserId::as_str);
45        let session_id_str = request.session_id.as_ref().map(SessionId::as_str);
46        let trace_id_str = request.trace_id.as_ref().map(TraceId::as_str);
47        let context_id_str = request.context_id.as_ref().map(ContextId::as_str);
48
49        sqlx::query_as!(
50            FileRow,
51            r#"
52            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)
53            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $12)
54            ON CONFLICT (path) DO UPDATE SET
55                public_url = EXCLUDED.public_url,
56                mime_type = EXCLUDED.mime_type,
57                size_bytes = EXCLUDED.size_bytes,
58                ai_content = EXCLUDED.ai_content,
59                metadata = EXCLUDED.metadata,
60                updated_at = EXCLUDED.updated_at
61            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
62            "#,
63            id_uuid,
64            request.path,
65            request.public_url,
66            request.mime_type,
67            request.size_bytes,
68            request.ai_content,
69            sqlx::types::Json(&request.metadata) as _,
70            user_id_str,
71            session_id_str,
72            trace_id_str,
73            context_id_str,
74            now
75        )
76        .fetch_one(&*self.write_pool)
77        .await?;
78
79        Ok(request.id)
80    }
81
82    pub async fn insert_file(&self, file: &File) -> FilesResult<FileId> {
83        let file_id = file.id.clone();
84
85        let mut request = InsertFileRequest::new(
86            file_id.clone(),
87            file.path.clone(),
88            file.public_url.clone(),
89            file.mime_type.clone(),
90        )
91        .with_ai_content(file.ai_content)
92        .with_metadata(file.metadata.0.clone());
93
94        if let Some(size) = file.size_bytes {
95            request = request.with_size(size);
96        }
97
98        if let Some(ref user_id) = file.user_id {
99            request = request.with_user_id(user_id.clone());
100        }
101
102        if let Some(ref session_id) = file.session_id {
103            request = request.with_session_id(session_id.clone());
104        }
105
106        if let Some(ref trace_id) = file.trace_id {
107            request = request.with_trace_id(trace_id.clone());
108        }
109
110        if let Some(ref context_id) = file.context_id {
111            request = request.with_context_id(context_id.clone());
112        }
113
114        self.insert(request).await
115    }
116
117    pub async fn find_by_id(&self, id: &FileId) -> FilesResult<Option<File>> {
118        let id_uuid = parse_file_uuid(id)?;
119
120        let result = sqlx::query_as!(
121            FileRow,
122            r#"
123            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
124            FROM files
125            WHERE id = $1 AND deleted_at IS NULL
126            "#,
127            id_uuid
128        )
129        .fetch_optional(&*self.pool)
130        .await?;
131
132        Ok(result.map(File::from))
133    }
134
135    pub async fn find_by_path(&self, path: &str) -> FilesResult<Option<File>> {
136        let result = sqlx::query_as!(
137            FileRow,
138            r#"
139            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
140            FROM files
141            WHERE path = $1 AND deleted_at IS NULL
142            "#,
143            path
144        )
145        .fetch_optional(&*self.pool)
146        .await?;
147
148        Ok(result.map(File::from))
149    }
150
151    pub async fn list_by_user(
152        &self,
153        user_id: &UserId,
154        limit: i64,
155        offset: i64,
156    ) -> FilesResult<Vec<File>> {
157        let user_id_str = user_id.as_str();
158        let result = sqlx::query_as!(
159            FileRow,
160            r#"
161            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
162            FROM files
163            WHERE user_id = $1 AND deleted_at IS NULL
164            ORDER BY created_at DESC
165            LIMIT $2 OFFSET $3
166            "#,
167            user_id_str,
168            limit,
169            offset
170        )
171        .fetch_all(&*self.pool)
172        .await?;
173
174        Ok(result.into_iter().map(File::from).collect())
175    }
176
177    pub async fn list_all(&self, limit: i64, offset: i64) -> FilesResult<Vec<File>> {
178        let result = sqlx::query_as!(
179            FileRow,
180            r#"
181            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
182            FROM files
183            WHERE deleted_at IS NULL
184            ORDER BY created_at DESC
185            LIMIT $1 OFFSET $2
186            "#,
187            limit,
188            offset
189        )
190        .fetch_all(&*self.pool)
191        .await?;
192
193        Ok(result.into_iter().map(File::from).collect())
194    }
195
196    pub async fn delete(&self, id: &FileId) -> FilesResult<()> {
197        let id_uuid = parse_file_uuid(id)?;
198
199        sqlx::query!(
200            r#"
201            DELETE FROM files
202            WHERE id = $1
203            "#,
204            id_uuid
205        )
206        .execute(&*self.write_pool)
207        .await?;
208
209        Ok(())
210    }
211
212    pub async fn update_metadata(&self, id: &FileId, metadata: &FileMetadata) -> FilesResult<()> {
213        let id_uuid = parse_file_uuid(id)?;
214        let now = Utc::now();
215
216        sqlx::query!(
217            r#"
218            UPDATE files
219            SET metadata = $1, updated_at = $2
220            WHERE id = $3
221            "#,
222            sqlx::types::Json(metadata) as _,
223            now,
224            id_uuid
225        )
226        .execute(&*self.write_pool)
227        .await?;
228
229        Ok(())
230    }
231
232    pub async fn search_by_path(&self, query: &str, limit: i64) -> FilesResult<Vec<File>> {
233        let pattern = format!("%{query}%");
234        let result = sqlx::query_as!(
235            FileRow,
236            r#"
237            SELECT id, path, public_url, mime_type, size_bytes, ai_content, metadata as "metadata: sqlx::types::Json<FileMetadata>",
238                   user_id as "user_id: UserId", session_id as "session_id: SessionId",
239                   trace_id as "trace_id: TraceId", context_id as "context_id: ContextId",
240                   created_at, updated_at, deleted_at
241            FROM files
242            WHERE path ILIKE $1 AND deleted_at IS NULL
243            ORDER BY created_at DESC
244            LIMIT $2
245            "#,
246            pattern,
247            limit
248        )
249        .fetch_all(&*self.pool)
250        .await?;
251
252        Ok(result.into_iter().map(File::from).collect())
253    }
254}
systemprompt_files/repository/file/mod.rs

systemprompt_files/repository/file/
mod.rs
