1use std::sync::Arc;
2
3use oxinbox_core::{Task, TaskStatus};
4use sqlx::PgPool;
5use tracing::instrument;
6use uuid::Uuid;
7
8use crate::repository::RepositoryError;
9
10pub struct ParadeDbRepository {
11 pool: PgPool,
12}
13
14impl ParadeDbRepository {
15 pub const fn new(pool: PgPool) -> Self {
16 Self { pool }
17 }
18
19 pub fn arc_new(pool: PgPool) -> Arc<Self> {
20 Arc::new(Self::new(pool))
21 }
22}
23
24impl ParadeDbRepository {
25 #[instrument(skip(self), fields(task_id = %task.id))]
26 pub async fn create(&self, task: &Task, user_id: i32) -> Result<Task, RepositoryError> {
27 sqlx::query_as::<_, TaskRow>(
28 r"INSERT INTO tasks (id, user_id, completed, priority, description, projects, contexts, status, created_at, updated_at, completed_at, due_date)
29 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)
30 RETURNING id, completed, priority, description, projects, contexts, status, created_at, updated_at, completed_at, due_date"
31 )
32 .bind(task.id)
33 .bind(user_id)
34 .bind(task.completed)
35 .bind(task.priority.map(|c| c.to_string()))
36 .bind(&task.description)
37 .bind(&task.projects)
38 .bind(&task.contexts)
39 .bind(format!("{:?}", task.status).to_lowercase())
40 .bind(task.created_at)
41 .bind(task.updated_at)
42 .bind(task.completed_at)
43 .bind(task.due_date)
44 .fetch_one(&self.pool)
45 .await
46 .map(TaskRow::into_task)
47 .map_err(RepositoryError::from)
48 }
49
50 #[instrument(skip(self), fields(task_id = %id))]
51 pub async fn get(&self, id: Uuid) -> Result<Task, RepositoryError> {
52 sqlx::query_as::<_, TaskRow>(
53 "SELECT id, completed, priority, description, projects, contexts, status, created_at, updated_at, completed_at, due_date FROM tasks WHERE id = $1"
54 )
55 .bind(id)
56 .fetch_optional(&self.pool)
57 .await
58 .map_err(RepositoryError::from)?
59 .map(TaskRow::into_task)
60 .ok_or(RepositoryError::NotFound(id))
61 }
62
63 #[instrument(skip(self))]
64 pub async fn list(&self, user_id: i32) -> Result<Vec<Task>, RepositoryError> {
65 let rows = sqlx::query_as::<_, TaskRow>(
66 "SELECT id, completed, priority, description, projects, contexts, status, created_at, updated_at, completed_at, due_date FROM tasks WHERE user_id = $1 ORDER BY created_at DESC"
67 )
68 .bind(user_id)
69 .fetch_all(&self.pool)
70 .await
71 .map_err(RepositoryError::from)?;
72
73 Ok(rows.into_iter().map(TaskRow::into_task).collect())
74 }
75
76 #[instrument(skip(self), fields(task_id = %task.id))]
77 pub async fn update(&self, task: &Task) -> Result<Task, RepositoryError> {
78 sqlx::query_as::<_, TaskRow>(
79 r"UPDATE tasks SET completed = $1, priority = $2, description = $3, projects = $4, contexts = $5, status = $6, updated_at = $7, completed_at = $8, due_date = $9
80 WHERE id = $10
81 RETURNING id, completed, priority, description, projects, contexts, status, created_at, updated_at, completed_at, due_date"
82 )
83 .bind(task.completed)
84 .bind(task.priority.map(|c| c.to_string()))
85 .bind(&task.description)
86 .bind(&task.projects)
87 .bind(&task.contexts)
88 .bind(format!("{:?}", task.status).to_lowercase())
89 .bind(task.updated_at)
90 .bind(task.completed_at)
91 .bind(task.due_date)
92 .bind(task.id)
93 .fetch_optional(&self.pool)
94 .await
95 .map_err(RepositoryError::from)?
96 .map(TaskRow::into_task)
97 .ok_or(RepositoryError::NotFound(task.id))
98 }
99
100 #[instrument(skip(self), fields(task_id = %id))]
101 pub async fn delete(&self, id: Uuid) -> Result<(), RepositoryError> {
102 let result = sqlx::query("DELETE FROM tasks WHERE id = $1")
103 .bind(id)
104 .execute(&self.pool)
105 .await
106 .map_err(RepositoryError::from)?;
107
108 if result.rows_affected() == 0 {
109 return Err(RepositoryError::NotFound(id));
110 }
111 Ok(())
112 }
113
114 #[instrument(skip(self))]
115 pub async fn search_bm25(
116 &self,
117 user_id: i32,
118 query: &str,
119 limit: i64,
120 ) -> Result<Vec<Task>, RepositoryError> {
121 let like = format!("%{query}%");
122 let rows = sqlx::query_as::<_, TaskRow>(
123 r"SELECT id, completed, priority, description, projects, contexts, status, created_at, updated_at, completed_at, due_date
124 FROM tasks
125 WHERE user_id = $1
126 AND (description ILIKE $2 OR $3 = ANY(projects) OR $4 = ANY(contexts))
127 ORDER BY created_at DESC
128 LIMIT $5"
129 )
130 .bind(user_id)
131 .bind(&like)
132 .bind(query)
133 .bind(query)
134 .bind(limit)
135 .fetch_all(&self.pool)
136 .await
137 .map_err(RepositoryError::from)?;
138
139 Ok(rows.into_iter().map(TaskRow::into_task).collect())
140 }
141
142 #[instrument(skip(self))]
143 pub async fn search_vector(
144 &self,
145 user_id: i32,
146 embedding: &[f32],
147 limit: i64,
148 ) -> Result<Vec<(Task, f64)>, RepositoryError> {
149 let vec_str = format!(
150 "[{}]",
151 embedding
152 .iter()
153 .map(ToString::to_string)
154 .collect::<Vec<_>>()
155 .join(",")
156 );
157 let rows = sqlx::query_as::<_, TaskRowWithScore>(
158 r"SELECT id, completed, priority, description, projects, contexts, status, created_at, updated_at, completed_at, due_date,
159 1 - (embedding <=> $1::vector) AS score
160 FROM tasks
161 WHERE user_id = $2 AND embedding IS NOT NULL
162 ORDER BY embedding <=> $1::vector
163 LIMIT $3"
164 )
165 .bind(&vec_str)
166 .bind(user_id)
167 .bind(limit)
168 .fetch_all(&self.pool)
169 .await
170 .map_err(RepositoryError::from)?;
171
172 Ok(rows
173 .into_iter()
174 .map(|r| -> (Task, f64) {
175 let score = r.score;
176 (r.into_task(), score)
177 })
178 .collect())
179 }
180
181 #[instrument(skip(self))]
182 pub async fn update_embedding(
183 &self,
184 task_id: Uuid,
185 embedding: &[f32],
186 ) -> Result<(), RepositoryError> {
187 let vec_str = format!(
188 "[{}]",
189 embedding
190 .iter()
191 .map(ToString::to_string)
192 .collect::<Vec<_>>()
193 .join(",")
194 );
195 sqlx::query("UPDATE tasks SET embedding = $1::vector WHERE id = $2")
196 .bind(&vec_str)
197 .bind(task_id)
198 .execute(&self.pool)
199 .await
200 .map_err(RepositoryError::from)?;
201 Ok(())
202 }
203
204 #[instrument(skip(self))]
205 pub async fn upsert_user(&self, email: &str) -> Result<i32, RepositoryError> {
206 let row: (i32,) = sqlx::query_as(
207 "INSERT INTO users (email) VALUES ($1) ON CONFLICT (email) DO UPDATE SET email = EXCLUDED.email RETURNING id"
208 )
209 .bind(email)
210 .fetch_one(&self.pool)
211 .await
212 .map_err(RepositoryError::from)?;
213 Ok(row.0)
214 }
215
216 #[instrument(skip(self))]
217 pub async fn create_session(&self, token: &str, user_id: i32) -> Result<(), RepositoryError> {
218 let expires = chrono::Utc::now() + chrono::TimeDelta::try_days(365).unwrap();
219 sqlx::query("INSERT INTO sessions (token, user_id, expires_at) VALUES ($1, $2, $3)")
220 .bind(token)
221 .bind(user_id)
222 .bind(expires)
223 .execute(&self.pool)
224 .await
225 .map_err(RepositoryError::from)?;
226 Ok(())
227 }
228
229 #[instrument(skip(self))]
230 pub async fn validate_session(&self, token: &str) -> Result<Option<i32>, RepositoryError> {
231 let row: Option<(i32,)> = sqlx::query_as(
232 "SELECT user_id FROM sessions WHERE token = $1 AND expires_at > CURRENT_TIMESTAMP",
233 )
234 .bind(token)
235 .fetch_optional(&self.pool)
236 .await
237 .map_err(RepositoryError::from)?;
238 Ok(row.map(|r| r.0))
239 }
240}
241
242#[derive(sqlx::FromRow)]
243struct TaskRowWithScore {
244 id: Uuid,
245 completed: bool,
246 priority: Option<String>,
247 description: String,
248 projects: Vec<String>,
249 contexts: Vec<String>,
250 status: String,
251 created_at: chrono::DateTime<chrono::Utc>,
252 updated_at: chrono::DateTime<chrono::Utc>,
253 completed_at: Option<chrono::DateTime<chrono::Utc>>,
254 due_date: Option<chrono::NaiveDate>,
255 score: f64,
256}
257
258impl TaskRowWithScore {
259 fn into_task(self) -> Task {
260 let status = match self.status.as_str() {
261 "inbox" => TaskStatus::Inbox,
262 "todo" => TaskStatus::Todo,
263 "doing" => TaskStatus::Doing,
264 "done" => TaskStatus::Done,
265 _ => TaskStatus::Someday,
266 };
267 Task {
268 id: self.id,
269 completed: self.completed,
270 priority: self
271 .priority
272 .and_then(|p| p.chars().next())
273 .filter(char::is_ascii_uppercase),
274 description: self.description,
275 projects: self.projects,
276 contexts: self.contexts,
277 status,
278 created_at: self.created_at,
279 updated_at: self.updated_at,
280 completed_at: self.completed_at,
281 due_date: self.due_date,
282 }
283 }
284}
285
286#[derive(sqlx::FromRow)]
287struct TaskRow {
288 id: Uuid,
289 completed: bool,
290 priority: Option<String>,
291 description: String,
292 projects: Vec<String>,
293 contexts: Vec<String>,
294 status: String,
295 created_at: chrono::DateTime<chrono::Utc>,
296 updated_at: chrono::DateTime<chrono::Utc>,
297 completed_at: Option<chrono::DateTime<chrono::Utc>>,
298 due_date: Option<chrono::NaiveDate>,
299}
300
301impl TaskRow {
302 fn into_task(self) -> Task {
303 let status = match self.status.as_str() {
304 "inbox" => TaskStatus::Inbox,
305 "todo" => TaskStatus::Todo,
306 "doing" => TaskStatus::Doing,
307 "done" => TaskStatus::Done,
308 _ => TaskStatus::Someday,
309 };
310 Task {
311 id: self.id,
312 completed: self.completed,
313 priority: self
314 .priority
315 .and_then(|p| p.chars().next())
316 .filter(char::is_ascii_uppercase),
317 description: self.description,
318 projects: self.projects,
319 contexts: self.contexts,
320 status,
321 created_at: self.created_at,
322 updated_at: self.updated_at,
323 completed_at: self.completed_at,
324 due_date: self.due_date,
325 }
326 }
327}