Tracing: DAO layer + handler spans on tracing, drop all Context params (#141)

trace_db_call no longer takes an OpenTelemetry Context: it opens a
db.<query_type>.<operation> span (same name via otel.name, same
db.query_type / db.operation attrs) on the caller's ambient tracing span
and runs the DAO body inside it. The cx / ctx / context parameter is gone
from every DAO trait method, impl and call site, and from the helpers and
background drains that only threaded it.

The remaining request-handler spans (files, tags, faces, image, video,
favorites, duplicates, memories) move to #[instrument]; the
Context::current_with_span / with_remote_span_context plumbing is gone, as
is extract_context_from_request (the tracing-actix-web root span covers
it). otel::set_span_attributes replaces BoxedSpan::set_attributes.

Background-job raw spans (thumbnails, video actors) are left for #142.

tests/otel_db_spans.rs pins name, attributes, status and nesting (across
an instrumented spawn); otel_traceparent now uses an #[instrument]
handler. CLAUDE.md documents the tracing conventions.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Cameron CordesandClaude Opus 5.5 committed 2026-09-29 17:02:40 -04:00
1 parent 1c508cdabd
commit b6deca2a52
55 files changed
+1942 -3852

No files matched your search

+15 -47
View File
@@ -46,23 +46,14 @@ pub struct InsertCalendarEvent {
pub trait CalendarEventDao: Sync + Send {
/// Store calendar event with optional embedding
fn store_event(
&mut self,
context: &opentelemetry::Context,
event: InsertCalendarEvent,
) -> Result<CalendarEvent, DbError>;
fn store_event(&mut self, event: InsertCalendarEvent) -> Result<CalendarEvent, DbError>;
/// Batch insert events (for import efficiency)
fn store_events_batch(
&mut self,
context: &opentelemetry::Context,
events: Vec<InsertCalendarEvent>,
) -> Result<usize, DbError>;
fn store_events_batch(&mut self, events: Vec<InsertCalendarEvent>) -> Result<usize, DbError>;
/// Find events in time range (PRIMARY query method)
fn find_events_in_range(
&mut self,
context: &opentelemetry::Context,
start_ts: i64,
end_ts: i64,
) -> Result<Vec<CalendarEvent>, DbError>;
@@ -70,7 +61,6 @@ pub trait CalendarEventDao: Sync + Send {
/// Find semantically similar events (SECONDARY - requires embeddings)
fn find_similar_events(
&mut self,
context: &opentelemetry::Context,
query_embedding: &[f32],
limit: usize,
) -> Result<Vec<CalendarEvent>, DbError>;
@@ -79,7 +69,6 @@ pub trait CalendarEventDao: Sync + Send {
/// "Events during photo timestamp ±N days, ranked by similarity to context"
fn find_relevant_events_hybrid(
&mut self,
context: &opentelemetry::Context,
center_timestamp: i64,
time_window_days: i64,
query_embedding: Option<&[f32]>,
@@ -87,15 +76,10 @@ pub trait CalendarEventDao: Sync + Send {
) -> Result<Vec<CalendarEvent>, DbError>;
/// Check if event exists (idempotency)
fn event_exists(
&mut self,
context: &opentelemetry::Context,
event_uid: &str,
start_time: i64,
) -> Result<bool, DbError>;
fn event_exists(&mut self, event_uid: &str, start_time: i64) -> Result<bool, DbError>;
/// Get count of events
fn get_event_count(&mut self, context: &opentelemetry::Context) -> Result<i64, DbError>;
fn get_event_count(&mut self) -> Result<i64, DbError>;
}
pub struct SqliteCalendarEventDao {
@@ -209,12 +193,8 @@ struct LastInsertRowId {
}
impl CalendarEventDao for SqliteCalendarEventDao {
fn store_event(
&mut self,
context: &opentelemetry::Context,
event: InsertCalendarEvent,
) -> Result<CalendarEvent, DbError> {
trace_db_call(context, "insert", "store_event", |_span| {
fn store_event(&mut self, event: InsertCalendarEvent) -> Result<CalendarEvent, DbError> {
trace_db_call("insert", "store_event", |_span| {
let mut conn = self
.connection
.lock()
@@ -278,12 +258,8 @@ impl CalendarEventDao for SqliteCalendarEventDao {
.map_err(|e| DbError::log(DbErrorKind::InsertError, e))
}
fn store_events_batch(
&mut self,
context: &opentelemetry::Context,
events: Vec<InsertCalendarEvent>,
) -> Result<usize, DbError> {
trace_db_call(context, "insert", "store_events_batch", |_span| {
fn store_events_batch(&mut self, events: Vec<InsertCalendarEvent>) -> Result<usize, DbError> {
trace_db_call("insert", "store_events_batch", |_span| {
let mut conn = self
.connection
.lock()
@@ -354,11 +330,10 @@ impl CalendarEventDao for SqliteCalendarEventDao {
fn find_events_in_range(
&mut self,
context: &opentelemetry::Context,
start_ts: i64,
end_ts: i64,
) -> Result<Vec<CalendarEvent>, DbError> {
trace_db_call(context, "query", "find_events_in_range", |_span| {
trace_db_call("query", "find_events_in_range", |_span| {
let mut conn = self.connection.lock().expect("Unable to get CalendarEventDao");
diesel::sql_query(
@@ -379,11 +354,10 @@ impl CalendarEventDao for SqliteCalendarEventDao {
fn find_similar_events(
&mut self,
context: &opentelemetry::Context,
query_embedding: &[f32],
limit: usize,
) -> Result<Vec<CalendarEvent>, DbError> {
trace_db_call(context, "query", "find_similar_events", |_span| {
trace_db_call("query", "find_similar_events", |_span| {
let mut conn = self.connection.lock().expect("Unable to get CalendarEventDao");
if query_embedding.len() != crate::ai::embedding_dim() {
@@ -436,13 +410,12 @@ impl CalendarEventDao for SqliteCalendarEventDao {
fn find_relevant_events_hybrid(
&mut self,
context: &opentelemetry::Context,
center_timestamp: i64,
time_window_days: i64,
query_embedding: Option<&[f32]>,
limit: usize,
) -> Result<Vec<CalendarEvent>, DbError> {
trace_db_call(context, "query", "find_relevant_events_hybrid", |_span| {
trace_db_call("query", "find_relevant_events_hybrid", |_span| {
let window_seconds = time_window_days * 86400;
let start_ts = center_timestamp - window_seconds;
let end_ts = center_timestamp + window_seconds;
@@ -506,13 +479,8 @@ impl CalendarEventDao for SqliteCalendarEventDao {
.map_err(|e| DbError::log(DbErrorKind::QueryError, e))
}
fn event_exists(
&mut self,
context: &opentelemetry::Context,
event_uid: &str,
start_time: i64,
) -> Result<bool, DbError> {
trace_db_call(context, "query", "event_exists", |_span| {
fn event_exists(&mut self, event_uid: &str, start_time: i64) -> Result<bool, DbError> {
trace_db_call("query", "event_exists", |_span| {
let mut conn = self.connection.lock().expect("Unable to get CalendarEventDao");
#[derive(QueryableByName)]
@@ -534,8 +502,8 @@ impl CalendarEventDao for SqliteCalendarEventDao {
.map_err(|e| DbError::log(DbErrorKind::QueryError, e))
}
fn get_event_count(&mut self, context: &opentelemetry::Context) -> Result<i64, DbError> {
trace_db_call(context, "query", "get_event_count", |_span| {
fn get_event_count(&mut self) -> Result<i64, DbError> {
trace_db_call("query", "get_event_count", |_span| {
let mut conn = self
.connection
.lock()
+14 -44
View File
@@ -36,16 +36,11 @@ pub struct InsertDailySummary {
pub trait DailySummaryDao: Sync + Send {
/// Store a daily summary with its embedding
fn store_summary(
&mut self,
context: &opentelemetry::Context,
summary: InsertDailySummary,
) -> Result<DailySummary, DbError>;
fn store_summary(&mut self, summary: InsertDailySummary) -> Result<DailySummary, DbError>;
/// Find semantically similar daily summaries using vector similarity
fn find_similar_summaries(
&mut self,
context: &opentelemetry::Context,
query_embedding: &[f32],
limit: usize,
) -> Result<Vec<DailySummary>, DbError>;
@@ -55,31 +50,21 @@ pub trait DailySummaryDao: Sync + Send {
/// Final score = similarity * time_weight, where time_weight decays with distance from target_date
fn find_similar_summaries_with_time_weight(
&mut self,
context: &opentelemetry::Context,
query_embedding: &[f32],
target_date: &str,
limit: usize,
) -> Result<Vec<DailySummary>, DbError>;
/// Check if a summary exists for a given date and contact
fn summary_exists(
&mut self,
context: &opentelemetry::Context,
date: &str,
contact: &str,
) -> Result<bool, DbError>;
fn summary_exists(&mut self, date: &str, contact: &str) -> Result<bool, DbError>;
/// Get count of summaries for a contact
fn get_summary_count(
&mut self,
context: &opentelemetry::Context,
contact: &str,
) -> Result<i64, DbError>;
fn get_summary_count(&mut self, contact: &str) -> Result<i64, DbError>;
/// Cheap presence check — returns true iff at least one daily summary row
/// exists. Used by gating logic that only needs "is the table empty?",
/// avoiding a `COUNT(*)` full scan on large corpora.
fn has_any_summaries(&mut self, context: &opentelemetry::Context) -> Result<bool, DbError>;
fn has_any_summaries(&mut self) -> Result<bool, DbError>;
}
pub struct SqliteDailySummaryDao {
@@ -138,12 +123,8 @@ impl SqliteDailySummaryDao {
}
impl DailySummaryDao for SqliteDailySummaryDao {
fn store_summary(
&mut self,
context: &opentelemetry::Context,
summary: InsertDailySummary,
) -> Result<DailySummary, DbError> {
trace_db_call(context, "insert", "store_summary", |_span| {
fn store_summary(&mut self, summary: InsertDailySummary) -> Result<DailySummary, DbError> {
trace_db_call("insert", "store_summary", |_span| {
let mut conn = self
.connection
.lock()
@@ -196,11 +177,10 @@ impl DailySummaryDao for SqliteDailySummaryDao {
fn find_similar_summaries(
&mut self,
context: &opentelemetry::Context,
query_embedding: &[f32],
limit: usize,
) -> Result<Vec<DailySummary>, DbError> {
trace_db_call(context, "query", "find_similar_summaries", |_span| {
trace_db_call("query", "find_similar_summaries", |_span| {
let mut conn = self.connection.lock().expect("Unable to get DailySummaryDao");
if query_embedding.len() != crate::ai::embedding_dim() {
@@ -293,12 +273,11 @@ impl DailySummaryDao for SqliteDailySummaryDao {
fn find_similar_summaries_with_time_weight(
&mut self,
context: &opentelemetry::Context,
query_embedding: &[f32],
target_date: &str,
limit: usize,
) -> Result<Vec<DailySummary>, DbError> {
trace_db_call(context, "query", "find_similar_summaries_with_time_weight", |_span| {
trace_db_call("query", "find_similar_summaries_with_time_weight", |_span| {
let mut conn = self.connection.lock().expect("Unable to get DailySummaryDao");
if query_embedding.len() != crate::ai::embedding_dim() {
@@ -414,13 +393,8 @@ impl DailySummaryDao for SqliteDailySummaryDao {
.map_err(|e| DbError::log(DbErrorKind::QueryError, e))
}
fn summary_exists(
&mut self,
context: &opentelemetry::Context,
date: &str,
contact: &str,
) -> Result<bool, DbError> {
trace_db_call(context, "query", "summary_exists", |_span| {
fn summary_exists(&mut self, date: &str, contact: &str) -> Result<bool, DbError> {
trace_db_call("query", "summary_exists", |_span| {
let mut conn = self
.connection
.lock()
@@ -441,12 +415,8 @@ impl DailySummaryDao for SqliteDailySummaryDao {
.map_err(|e| DbError::log(DbErrorKind::QueryError, e))
}
fn get_summary_count(
&mut self,
context: &opentelemetry::Context,
contact: &str,
) -> Result<i64, DbError> {
trace_db_call(context, "query", "get_summary_count", |_span| {
fn get_summary_count(&mut self, contact: &str) -> Result<i64, DbError> {
trace_db_call("query", "get_summary_count", |_span| {
let mut conn = self
.connection
.lock()
@@ -463,8 +433,8 @@ impl DailySummaryDao for SqliteDailySummaryDao {
.map_err(|e| DbError::log(DbErrorKind::QueryError, e))
}
fn has_any_summaries(&mut self, context: &opentelemetry::Context) -> Result<bool, DbError> {
trace_db_call(context, "query", "has_any_summaries", |_span| {
fn has_any_summaries(&mut self) -> Result<bool, DbError> {
trace_db_call("query", "has_any_summaries", |_span| {
let mut conn = self
.connection
.lock()
+65 -134
View File
@@ -18,7 +18,6 @@ pub trait InsightGenerationJobDao: Sync + Send {
/// Cleans up terminal-state rows for the same key first.
fn create_job(
&mut self,
context: &opentelemetry::Context,
library_id: i32,
file_path: &str,
generation_type: InsightGenerationType,
@@ -27,59 +26,34 @@ pub trait InsightGenerationJobDao: Sync + Send {
/// Mark a job as completed with the resulting insight id. Only updates
/// if the job is still in "running" status (prevents overwriting a
/// cancelled job with a late-completing task).
fn complete_job(
&mut self,
context: &opentelemetry::Context,
job_id: i32,
insight_id: i32,
) -> Result<(), DbError>;
fn complete_job(&mut self, job_id: i32, insight_id: i32) -> Result<(), DbError>;
/// Mark a job as failed with an error message. Only updates if the job
/// is still in "running" status.
fn fail_job(
&mut self,
context: &opentelemetry::Context,
job_id: i32,
error_message: &str,
) -> Result<(), DbError>;
fn fail_job(&mut self, job_id: i32, error_message: &str) -> Result<(), DbError>;
/// Cancel a specific job by id. Only updates if the job is still
/// in "running" status. Returns true if a row was updated.
fn cancel_job(
&mut self,
context: &opentelemetry::Context,
job_id: i32,
) -> Result<bool, DbError>;
fn cancel_job(&mut self, job_id: i32) -> Result<bool, DbError>;
/// Cancel all running jobs for a given file. Returns the number of
/// jobs cancelled.
fn cancel_active_jobs(
&mut self,
context: &opentelemetry::Context,
library_id: i32,
file_path: &str,
) -> Result<usize, DbError>;
fn cancel_active_jobs(&mut self, library_id: i32, file_path: &str) -> Result<usize, DbError>;
/// Find the latest running job for a given file. Returns None if no
/// running job exists.
fn get_active_job(
&mut self,
context: &opentelemetry::Context,
library_id: i32,
file_path: &str,
) -> Result<Option<InsightGenerationJob>, DbError>;
/// Find any job by id regardless of status.
fn get_job_by_id(
&mut self,
context: &opentelemetry::Context,
job_id: i32,
) -> Result<Option<InsightGenerationJob>, DbError>;
fn get_job_by_id(&mut self, job_id: i32) -> Result<Option<InsightGenerationJob>, DbError>;
/// Mark all jobs still in "running" status as "failed" with a recovery
/// error message. Returns the number of jobs recovered.
fn recover_orphaned_jobs(&mut self, context: &opentelemetry::Context)
-> Result<usize, DbError>;
fn recover_orphaned_jobs(&mut self) -> Result<usize, DbError>;
}
pub struct SqliteInsightGenerationJobDao {
@@ -108,12 +82,11 @@ impl SqliteInsightGenerationJobDao {
impl InsightGenerationJobDao for SqliteInsightGenerationJobDao {
fn create_job(
&mut self,
context: &opentelemetry::Context,
library_id: i32,
file_path: &str,
generation_type: InsightGenerationType,
) -> Result<i32, DbError> {
trace_db_call(context, "insert", "create_job", |_span| {
trace_db_call("insert", "create_job", |_span| {
use schema::insight_generation_jobs::dsl;
let mut connection = self
@@ -155,13 +128,8 @@ impl InsightGenerationJobDao for SqliteInsightGenerationJobDao {
.map_err(|e| DbError::log(DbErrorKind::QueryError, e))
}
fn complete_job(
&mut self,
context: &opentelemetry::Context,
job_id: i32,
insight_id: i32,
) -> Result<(), DbError> {
trace_db_call(context, "update", "complete_job", |_span| {
fn complete_job(&mut self, job_id: i32, insight_id: i32) -> Result<(), DbError> {
trace_db_call("update", "complete_job", |_span| {
use schema::insight_generation_jobs::dsl;
let mut connection = self
@@ -195,13 +163,8 @@ impl InsightGenerationJobDao for SqliteInsightGenerationJobDao {
.map_err(|e| DbError::log(DbErrorKind::UpdateError, e))
}
fn fail_job(
&mut self,
context: &opentelemetry::Context,
job_id: i32,
error_message: &str,
) -> Result<(), DbError> {
trace_db_call(context, "update", "fail_job", |_span| {
fn fail_job(&mut self, job_id: i32, error_message: &str) -> Result<(), DbError> {
trace_db_call("update", "fail_job", |_span| {
use schema::insight_generation_jobs::dsl;
let mut connection = self
@@ -234,12 +197,8 @@ impl InsightGenerationJobDao for SqliteInsightGenerationJobDao {
.map_err(|e| DbError::log(DbErrorKind::UpdateError, e))
}
fn cancel_job(
&mut self,
context: &opentelemetry::Context,
job_id: i32,
) -> Result<bool, DbError> {
trace_db_call(context, "update", "cancel_job", |_span| {
fn cancel_job(&mut self, job_id: i32) -> Result<bool, DbError> {
trace_db_call("update", "cancel_job", |_span| {
use schema::insight_generation_jobs::dsl;
let mut connection = self
@@ -272,13 +231,8 @@ impl InsightGenerationJobDao for SqliteInsightGenerationJobDao {
.map_err(|e| DbError::log(DbErrorKind::UpdateError, e))
}
fn cancel_active_jobs(
&mut self,
context: &opentelemetry::Context,
library_id: i32,
file_path: &str,
) -> Result<usize, DbError> {
trace_db_call(context, "update", "cancel_active_jobs", |_span| {
fn cancel_active_jobs(&mut self, library_id: i32, file_path: &str) -> Result<usize, DbError> {
trace_db_call("update", "cancel_active_jobs", |_span| {
use schema::insight_generation_jobs::dsl;
let mut connection = self
@@ -314,11 +268,10 @@ impl InsightGenerationJobDao for SqliteInsightGenerationJobDao {
fn get_active_job(
&mut self,
context: &opentelemetry::Context,
library_id: i32,
file_path: &str,
) -> Result<Option<InsightGenerationJob>, DbError> {
trace_db_call(context, "query", "get_active_job", |_span| {
trace_db_call("query", "get_active_job", |_span| {
use schema::insight_generation_jobs::dsl;
let mut connection = self
@@ -341,12 +294,8 @@ impl InsightGenerationJobDao for SqliteInsightGenerationJobDao {
.map_err(|e| DbError::log(DbErrorKind::QueryError, e))
}
fn get_job_by_id(
&mut self,
context: &opentelemetry::Context,
job_id: i32,
) -> Result<Option<InsightGenerationJob>, DbError> {
trace_db_call(context, "query", "get_job_by_id", |_span| {
fn get_job_by_id(&mut self, job_id: i32) -> Result<Option<InsightGenerationJob>, DbError> {
trace_db_call("query", "get_job_by_id", |_span| {
use schema::insight_generation_jobs::dsl;
let mut connection = self
@@ -363,11 +312,8 @@ impl InsightGenerationJobDao for SqliteInsightGenerationJobDao {
.map_err(|e| DbError::log(DbErrorKind::QueryError, e))
}
fn recover_orphaned_jobs(
&mut self,
context: &opentelemetry::Context,
) -> Result<usize, DbError> {
trace_db_call(context, "update", "recover_orphaned_jobs", |_span| {
fn recover_orphaned_jobs(&mut self) -> Result<usize, DbError> {
trace_db_call("update", "recover_orphaned_jobs", |_span| {
use schema::insight_generation_jobs::dsl;
let mut connection = self
@@ -414,21 +360,16 @@ mod tests {
SqliteInsightGenerationJobDao::from_connection(Arc::new(Mutex::new(conn)))
}
fn ctx() -> opentelemetry::Context {
opentelemetry::Context::new()
}
#[test]
fn create_job_inserts_new_row() {
let mut dao = setup_dao();
let ctx = ctx();
let job_id_1 = dao
.create_job(&ctx, 1, "photos/test.jpg", InsightGenerationType::Standard)
.create_job(1, "photos/test.jpg", InsightGenerationType::Standard)
.unwrap();
let job_id_2 = dao
.create_job(&ctx, 1, "photos/test.jpg", InsightGenerationType::Standard)
.create_job(1, "photos/test.jpg", InsightGenerationType::Standard)
.unwrap();
assert_ne!(job_id_1, job_id_2, "each create_job call inserts a new row");
@@ -437,15 +378,14 @@ mod tests {
#[test]
fn complete_job_sets_result() {
let mut dao = setup_dao();
let ctx = ctx();
let job_id = dao
.create_job(&ctx, 1, "photos/test.jpg", InsightGenerationType::Standard)
.create_job(1, "photos/test.jpg", InsightGenerationType::Standard)
.unwrap();
dao.complete_job(&ctx, job_id, 42).unwrap();
dao.complete_job(job_id, 42).unwrap();
let job = dao.get_job_by_id(&ctx, job_id).unwrap().unwrap();
let job = dao.get_job_by_id(job_id).unwrap().unwrap();
assert_eq!(job.status, InsightJobStatus::Completed.as_str());
assert_eq!(job.result_insight_id, Some(42));
assert!(job.completed_at.is_some());
@@ -454,15 +394,14 @@ mod tests {
#[test]
fn fail_job_sets_error() {
let mut dao = setup_dao();
let ctx = ctx();
let job_id = dao
.create_job(&ctx, 1, "photos/test.jpg", InsightGenerationType::Agentic)
.create_job(1, "photos/test.jpg", InsightGenerationType::Agentic)
.unwrap();
dao.fail_job(&ctx, job_id, "model timeout").unwrap();
dao.fail_job(job_id, "model timeout").unwrap();
let job = dao.get_job_by_id(&ctx, job_id).unwrap().unwrap();
let job = dao.get_job_by_id(job_id).unwrap().unwrap();
assert_eq!(job.status, InsightJobStatus::Failed.as_str());
assert_eq!(job.error_message.as_deref(), Some("model timeout"));
assert!(job.completed_at.is_some());
@@ -471,61 +410,58 @@ mod tests {
#[test]
fn get_active_job_returns_none_when_completed() {
let mut dao = setup_dao();
let ctx = ctx();
let job_id = dao
.create_job(&ctx, 1, "photos/test.jpg", InsightGenerationType::Standard)
.create_job(1, "photos/test.jpg", InsightGenerationType::Standard)
.unwrap();
// Job is running
let active = dao.get_active_job(&ctx, 1, "photos/test.jpg").unwrap();
let active = dao.get_active_job(1, "photos/test.jpg").unwrap();
assert!(active.is_some());
assert_eq!(active.unwrap().id, job_id);
// Complete it
dao.complete_job(&ctx, job_id, 1).unwrap();
dao.complete_job(job_id, 1).unwrap();
// No longer active
let active = dao.get_active_job(&ctx, 1, "photos/test.jpg").unwrap();
let active = dao.get_active_job(1, "photos/test.jpg").unwrap();
assert!(active.is_none());
}
#[test]
fn cancel_active_jobs() {
let mut dao = setup_dao();
let ctx = ctx();
let job_id = dao
.create_job(&ctx, 1, "photos/test.jpg", InsightGenerationType::Standard)
.create_job(1, "photos/test.jpg", InsightGenerationType::Standard)
.unwrap();
let cancelled = dao.cancel_active_jobs(&ctx, 1, "photos/test.jpg").unwrap();
let cancelled = dao.cancel_active_jobs(1, "photos/test.jpg").unwrap();
assert_eq!(cancelled, 1, "should cancel 1 running job");
// Job is no longer active
let active = dao.get_active_job(&ctx, 1, "photos/test.jpg").unwrap();
let active = dao.get_active_job(1, "photos/test.jpg").unwrap();
assert!(active.is_none());
// Job exists with cancelled status
let job = dao.get_job_by_id(&ctx, job_id).unwrap().unwrap();
let job = dao.get_job_by_id(job_id).unwrap().unwrap();
assert_eq!(job.status, InsightJobStatus::Cancelled.as_str());
// Cancelling again returns 0 (nothing to cancel)
let cancelled2 = dao.cancel_active_jobs(&ctx, 1, "photos/test.jpg").unwrap();
let cancelled2 = dao.cancel_active_jobs(1, "photos/test.jpg").unwrap();
assert_eq!(cancelled2, 0, "should return 0 when no running job");
}
#[test]
fn get_active_job_scoped_by_library() {
let mut dao = setup_dao();
let ctx = ctx();
let job_id_1 = dao
.create_job(&ctx, 1, "photos/test.jpg", InsightGenerationType::Standard)
.create_job(1, "photos/test.jpg", InsightGenerationType::Standard)
.unwrap();
let job_id_2 = dao
.create_job(&ctx, 2, "photos/test.jpg", InsightGenerationType::Standard)
.create_job(2, "photos/test.jpg", InsightGenerationType::Standard)
.unwrap();
assert_ne!(
@@ -534,14 +470,14 @@ mod tests {
);
// Complete lib1's job
dao.complete_job(&ctx, job_id_1, 1).unwrap();
dao.complete_job(job_id_1, 1).unwrap();
// lib1 has no active job
let active1 = dao.get_active_job(&ctx, 1, "photos/test.jpg").unwrap();
let active1 = dao.get_active_job(1, "photos/test.jpg").unwrap();
assert!(active1.is_none());
// lib2 still has active job
let active2 = dao.get_active_job(&ctx, 2, "photos/test.jpg").unwrap();
let active2 = dao.get_active_job(2, "photos/test.jpg").unwrap();
assert!(active2.is_some());
assert_eq!(active2.unwrap().id, job_id_2);
}
@@ -549,21 +485,20 @@ mod tests {
#[test]
fn get_job_by_id_finds_any_status() {
let mut dao = setup_dao();
let ctx = ctx();
let job_id = dao
.create_job(&ctx, 1, "photos/test.jpg", InsightGenerationType::Standard)
.create_job(1, "photos/test.jpg", InsightGenerationType::Standard)
.unwrap();
// Find while running
let job = dao.get_job_by_id(&ctx, job_id).unwrap().unwrap();
let job = dao.get_job_by_id(job_id).unwrap().unwrap();
assert_eq!(job.status, InsightJobStatus::Running.as_str());
// Complete it
dao.complete_job(&ctx, job_id, 99).unwrap();
dao.complete_job(job_id, 99).unwrap();
// Still findable
let job = dao.get_job_by_id(&ctx, job_id).unwrap().unwrap();
let job = dao.get_job_by_id(job_id).unwrap().unwrap();
assert_eq!(job.status, InsightJobStatus::Completed.as_str());
assert_eq!(job.result_insight_id, Some(99));
}
@@ -571,29 +506,28 @@ mod tests {
#[test]
fn recover_orphaned_jobs() {
let mut dao = setup_dao();
let ctx = ctx();
// Create two running jobs
let job_id_1 = dao
.create_job(&ctx, 1, "photos/a.jpg", InsightGenerationType::Standard)
.create_job(1, "photos/a.jpg", InsightGenerationType::Standard)
.unwrap();
let job_id_2 = dao
.create_job(&ctx, 1, "photos/b.jpg", InsightGenerationType::Agentic)
.create_job(1, "photos/b.jpg", InsightGenerationType::Agentic)
.unwrap();
// Complete one
dao.complete_job(&ctx, job_id_1, 1).unwrap();
dao.complete_job(job_id_1, 1).unwrap();
// Recover should only affect the running job
let recovered = dao.recover_orphaned_jobs(&ctx).unwrap();
let recovered = dao.recover_orphaned_jobs().unwrap();
assert_eq!(recovered, 1, "should recover exactly 1 running job");
// job_id_1 is still completed
let job1 = dao.get_job_by_id(&ctx, job_id_1).unwrap().unwrap();
let job1 = dao.get_job_by_id(job_id_1).unwrap().unwrap();
assert_eq!(job1.status, InsightJobStatus::Completed.as_str());
// job_id_2 is now failed with recovery message
let job2 = dao.get_job_by_id(&ctx, job_id_2).unwrap().unwrap();
let job2 = dao.get_job_by_id(job_id_2).unwrap().unwrap();
assert_eq!(job2.status, InsightJobStatus::Failed.as_str());
assert_eq!(
job2.error_message.as_deref(),
@@ -601,25 +535,24 @@ mod tests {
);
// Second recovery is a no-op
let recovered2 = dao.recover_orphaned_jobs(&ctx).unwrap();
let recovered2 = dao.recover_orphaned_jobs().unwrap();
assert_eq!(recovered2, 0, "no running jobs remain");
}
#[test]
fn complete_job_noop_when_cancelled() {
let mut dao = setup_dao();
let ctx = ctx();
let job_id = dao
.create_job(&ctx, 1, "photos/test.jpg", InsightGenerationType::Standard)
.create_job(1, "photos/test.jpg", InsightGenerationType::Standard)
.unwrap();
dao.cancel_job(&ctx, job_id).unwrap();
dao.cancel_job(job_id).unwrap();
// Late-completing task tries to mark as completed — should be a no-op
dao.complete_job(&ctx, job_id, 42).unwrap();
dao.complete_job(job_id, 42).unwrap();
let job = dao.get_job_by_id(&ctx, job_id).unwrap().unwrap();
let job = dao.get_job_by_id(job_id).unwrap().unwrap();
assert_eq!(
job.status,
InsightJobStatus::Cancelled.as_str(),
@@ -634,18 +567,17 @@ mod tests {
#[test]
fn fail_job_noop_when_cancelled() {
let mut dao = setup_dao();
let ctx = ctx();
let job_id = dao
.create_job(&ctx, 1, "photos/test.jpg", InsightGenerationType::Agentic)
.create_job(1, "photos/test.jpg", InsightGenerationType::Agentic)
.unwrap();
dao.cancel_job(&ctx, job_id).unwrap();
dao.cancel_job(job_id).unwrap();
// Late-failing task tries to mark as failed — should be a no-op
dao.fail_job(&ctx, job_id, "timeout after 120s").unwrap();
dao.fail_job(job_id, "timeout after 120s").unwrap();
let job = dao.get_job_by_id(&ctx, job_id).unwrap().unwrap();
let job = dao.get_job_by_id(job_id).unwrap().unwrap();
assert_eq!(
job.status,
InsightJobStatus::Cancelled.as_str(),
@@ -661,21 +593,20 @@ mod tests {
#[test]
fn cancel_job_by_id() {
let mut dao = setup_dao();
let ctx = ctx();
let job_id = dao
.create_job(&ctx, 1, "photos/test.jpg", InsightGenerationType::Standard)
.create_job(1, "photos/test.jpg", InsightGenerationType::Standard)
.unwrap();
let cancelled = dao.cancel_job(&ctx, job_id).unwrap();
let cancelled = dao.cancel_job(job_id).unwrap();
assert!(cancelled, "should cancel running job");
let job = dao.get_job_by_id(&ctx, job_id).unwrap().unwrap();
let job = dao.get_job_by_id(job_id).unwrap().unwrap();
assert_eq!(job.status, InsightJobStatus::Cancelled.as_str());
assert!(job.completed_at.is_some());
// Cancelling again is a no-op
let cancelled2 = dao.cancel_job(&ctx, job_id).unwrap();
let cancelled2 = dao.cancel_job(job_id).unwrap();
assert!(!cancelled2, "already cancelled job should return false");
}
}
+52 -144
View File
@@ -9,17 +9,9 @@ use crate::database::{DbError, DbErrorKind, connect};
use crate::otel::trace_db_call;
pub trait InsightDao: Sync + Send {
fn store_insight(
&mut self,
context: &opentelemetry::Context,
insight: InsertPhotoInsight,
) -> Result<PhotoInsight, DbError>;
fn store_insight(&mut self, insight: InsertPhotoInsight) -> Result<PhotoInsight, DbError>;
fn get_insight(
&mut self,
context: &opentelemetry::Context,
file_path: &str,
) -> Result<Option<PhotoInsight>, DbError>;
fn get_insight(&mut self, file_path: &str) -> Result<Option<PhotoInsight>, DbError>;
/// Library-scoped variant of `get_insight`. The default `get_insight`
/// finds any `is_current=true` row matching `file_path` across
@@ -32,7 +24,6 @@ pub trait InsightDao: Sync + Send {
/// one (chat / insight write paths always do).
fn get_current_insight_for_library(
&mut self,
context: &opentelemetry::Context,
library_id: i32,
file_path: &str,
) -> Result<Option<PhotoInsight>, DbError>;
@@ -41,61 +32,29 @@ pub trait InsightDao: Sync + Send {
/// `paths`. Used for content-hash sharing: the caller expands a
/// single file into all rel_paths with the same content_hash, then
/// asks here for any existing insight attached to any of them.
fn get_insight_for_paths(
&mut self,
context: &opentelemetry::Context,
paths: &[String],
) -> Result<Option<PhotoInsight>, DbError>;
fn get_insight_for_paths(&mut self, paths: &[String]) -> Result<Option<PhotoInsight>, DbError>;
fn get_insight_history(
&mut self,
context: &opentelemetry::Context,
file_path: &str,
) -> Result<Vec<PhotoInsight>, DbError>;
fn get_insight_history(&mut self, file_path: &str) -> Result<Vec<PhotoInsight>, DbError>;
/// Fetch a single insight by primary key, regardless of `is_current`.
/// Used by the few-shot injection flow where the caller picks specific
/// historical insights (which may have been superseded) as training
/// exemplars for a fresh generation.
fn get_insight_by_id(
&mut self,
context: &opentelemetry::Context,
insight_id: i32,
) -> Result<Option<PhotoInsight>, DbError>;
fn get_insight_by_id(&mut self, insight_id: i32) -> Result<Option<PhotoInsight>, DbError>;
fn delete_insight(
&mut self,
context: &opentelemetry::Context,
file_path: &str,
) -> Result<(), DbError>;
fn delete_insight(&mut self, file_path: &str) -> Result<(), DbError>;
fn get_all_insights(
&mut self,
context: &opentelemetry::Context,
) -> Result<Vec<PhotoInsight>, DbError>;
fn get_all_insights(&mut self) -> Result<Vec<PhotoInsight>, DbError>;
fn rate_insight(
&mut self,
context: &opentelemetry::Context,
file_path: &str,
approved: bool,
) -> Result<(), DbError>;
fn rate_insight(&mut self, file_path: &str, approved: bool) -> Result<(), DbError>;
/// Rate a specific insight version by primary key, regardless of
/// `is_current`. Used by the per-file history view to approve/reject
/// previously generated (superseded) versions, which the path-based
/// `rate_insight` (current row only) cannot reach.
fn rate_insight_by_id(
&mut self,
context: &opentelemetry::Context,
insight_id: i32,
approved: bool,
) -> Result<(), DbError>;
fn rate_insight_by_id(&mut self, insight_id: i32, approved: bool) -> Result<(), DbError>;
fn get_approved_insights(
&mut self,
context: &opentelemetry::Context,
) -> Result<Vec<PhotoInsight>, DbError>;
fn get_approved_insights(&mut self) -> Result<Vec<PhotoInsight>, DbError>;
/// Replace the `training_messages` JSON blob on the current row for
/// `(library_id, rel_path)`. Used by chat-turn append mode to persist
@@ -104,7 +63,6 @@ pub trait InsightDao: Sync + Send {
/// indicating a concurrent regenerate/reconcile flipped `is_current`).
fn update_training_messages(
&mut self,
context: &opentelemetry::Context,
library_id: i32,
file_path: &str,
training_messages_json: &str,
@@ -136,12 +94,8 @@ impl SqliteInsightDao {
}
impl InsightDao for SqliteInsightDao {
fn store_insight(
&mut self,
context: &opentelemetry::Context,
mut insight: InsertPhotoInsight,
) -> Result<PhotoInsight, DbError> {
trace_db_call(context, "insert", "store_insight", |_span| {
fn store_insight(&mut self, mut insight: InsertPhotoInsight) -> Result<PhotoInsight, DbError> {
trace_db_call("insert", "store_insight", |_span| {
use schema::photo_insights::dsl::*;
let mut connection = self.connection.lock().expect("Unable to get InsightDao");
@@ -193,12 +147,8 @@ impl InsightDao for SqliteInsightDao {
})
}
fn get_insight(
&mut self,
context: &opentelemetry::Context,
path: &str,
) -> Result<Option<PhotoInsight>, DbError> {
trace_db_call(context, "query", "get_insight", |_span| {
fn get_insight(&mut self, path: &str) -> Result<Option<PhotoInsight>, DbError> {
trace_db_call("query", "get_insight", |_span| {
use schema::photo_insights::dsl::*;
let mut connection = self.connection.lock().expect("Unable to get InsightDao");
@@ -215,40 +165,30 @@ impl InsightDao for SqliteInsightDao {
fn get_current_insight_for_library(
&mut self,
context: &opentelemetry::Context,
lib_id: i32,
path: &str,
) -> Result<Option<PhotoInsight>, DbError> {
trace_db_call(
context,
"query",
"get_current_insight_for_library",
|_span| {
use schema::photo_insights::dsl::*;
trace_db_call("query", "get_current_insight_for_library", |_span| {
use schema::photo_insights::dsl::*;
let mut connection = self.connection.lock().expect("Unable to get InsightDao");
let mut connection = self.connection.lock().expect("Unable to get InsightDao");
photo_insights
.filter(library_id.eq(lib_id))
.filter(rel_path.eq(path))
.filter(is_current.eq(true))
.first::<PhotoInsight>(connection.deref_mut())
.optional()
.map_err(|e| anyhow::anyhow!("Query error: {}", e))
},
)
photo_insights
.filter(library_id.eq(lib_id))
.filter(rel_path.eq(path))
.filter(is_current.eq(true))
.first::<PhotoInsight>(connection.deref_mut())
.optional()
.map_err(|e| anyhow::anyhow!("Query error: {}", e))
})
.map_err(|e| DbError::log(DbErrorKind::QueryError, e))
}
fn get_insight_for_paths(
&mut self,
context: &opentelemetry::Context,
paths: &[String],
) -> Result<Option<PhotoInsight>, DbError> {
fn get_insight_for_paths(&mut self, paths: &[String]) -> Result<Option<PhotoInsight>, DbError> {
if paths.is_empty() {
return Ok(None);
}
trace_db_call(context, "query", "get_insight_for_paths", |_span| {
trace_db_call("query", "get_insight_for_paths", |_span| {
use schema::photo_insights::dsl::*;
let mut connection = self.connection.lock().expect("Unable to get InsightDao");
@@ -264,12 +204,8 @@ impl InsightDao for SqliteInsightDao {
.map_err(|e| DbError::log(DbErrorKind::QueryError, e))
}
fn get_insight_history(
&mut self,
context: &opentelemetry::Context,
path: &str,
) -> Result<Vec<PhotoInsight>, DbError> {
trace_db_call(context, "query", "get_insight_history", |_span| {
fn get_insight_history(&mut self, path: &str) -> Result<Vec<PhotoInsight>, DbError> {
trace_db_call("query", "get_insight_history", |_span| {
use schema::photo_insights::dsl::*;
let mut connection = self.connection.lock().expect("Unable to get InsightDao");
@@ -283,12 +219,8 @@ impl InsightDao for SqliteInsightDao {
.map_err(|e| DbError::log(DbErrorKind::QueryError, e))
}
fn get_insight_by_id(
&mut self,
context: &opentelemetry::Context,
insight_id: i32,
) -> Result<Option<PhotoInsight>, DbError> {
trace_db_call(context, "query", "get_insight_by_id", |_span| {
fn get_insight_by_id(&mut self, insight_id: i32) -> Result<Option<PhotoInsight>, DbError> {
trace_db_call("query", "get_insight_by_id", |_span| {
use schema::photo_insights::dsl::*;
let mut connection = self.connection.lock().expect("Unable to get InsightDao");
@@ -302,12 +234,8 @@ impl InsightDao for SqliteInsightDao {
.map_err(|e| DbError::log(DbErrorKind::QueryError, e))
}
fn delete_insight(
&mut self,
context: &opentelemetry::Context,
path: &str,
) -> Result<(), DbError> {
trace_db_call(context, "delete", "delete_insight", |_span| {
fn delete_insight(&mut self, path: &str) -> Result<(), DbError> {
trace_db_call("delete", "delete_insight", |_span| {
use schema::photo_insights::dsl::*;
let mut connection = self.connection.lock().expect("Unable to get InsightDao");
@@ -320,11 +248,8 @@ impl InsightDao for SqliteInsightDao {
.map_err(|e| DbError::log(DbErrorKind::QueryError, e))
}
fn get_all_insights(
&mut self,
context: &opentelemetry::Context,
) -> Result<Vec<PhotoInsight>, DbError> {
trace_db_call(context, "query", "get_all_insights", |_span| {
fn get_all_insights(&mut self) -> Result<Vec<PhotoInsight>, DbError> {
trace_db_call("query", "get_all_insights", |_span| {
use schema::photo_insights::dsl::*;
let mut connection = self.connection.lock().expect("Unable to get InsightDao");
@@ -338,13 +263,8 @@ impl InsightDao for SqliteInsightDao {
.map_err(|e| DbError::log(DbErrorKind::QueryError, e))
}
fn rate_insight(
&mut self,
context: &opentelemetry::Context,
path: &str,
is_approved: bool,
) -> Result<(), DbError> {
trace_db_call(context, "update", "rate_insight", |_span| {
fn rate_insight(&mut self, path: &str, is_approved: bool) -> Result<(), DbError> {
trace_db_call("update", "rate_insight", |_span| {
use schema::photo_insights::dsl::*;
let mut connection = self.connection.lock().expect("Unable to get InsightDao");
@@ -362,13 +282,8 @@ impl InsightDao for SqliteInsightDao {
.map_err(|e| DbError::log(DbErrorKind::UpdateError, e))
}
fn rate_insight_by_id(
&mut self,
context: &opentelemetry::Context,
target_id: i32,
is_approved: bool,
) -> Result<(), DbError> {
trace_db_call(context, "update", "rate_insight_by_id", |_span| {
fn rate_insight_by_id(&mut self, target_id: i32, is_approved: bool) -> Result<(), DbError> {
trace_db_call("update", "rate_insight_by_id", |_span| {
use schema::photo_insights::dsl::*;
let mut connection = self.connection.lock().expect("Unable to get InsightDao");
@@ -382,11 +297,8 @@ impl InsightDao for SqliteInsightDao {
.map_err(|e| DbError::log(DbErrorKind::UpdateError, e))
}
fn get_approved_insights(
&mut self,
context: &opentelemetry::Context,
) -> Result<Vec<PhotoInsight>, DbError> {
trace_db_call(context, "query", "get_approved_insights", |_span| {
fn get_approved_insights(&mut self) -> Result<Vec<PhotoInsight>, DbError> {
trace_db_call("query", "get_approved_insights", |_span| {
use schema::photo_insights::dsl::*;
let mut connection = self.connection.lock().expect("Unable to get InsightDao");
@@ -403,12 +315,11 @@ impl InsightDao for SqliteInsightDao {
fn update_training_messages(
&mut self,
context: &opentelemetry::Context,
lib_id: i32,
path: &str,
training_messages_json: &str,
) -> Result<usize, DbError> {
trace_db_call(context, "update", "update_training_messages", |_span| {
trace_db_call("update", "update_training_messages", |_span| {
use schema::photo_insights::dsl::*;
let mut connection = self.connection.lock().expect("Unable to get InsightDao");
@@ -466,19 +377,17 @@ mod tests {
#[test]
fn get_insight_history_returns_all_versions_newest_first() {
let cx = opentelemetry::Context::new();
let mut dao = dao();
// store_insight flips prior rows to is_current=false, so three
// generations for the same path leave a 3-row history.
dao.store_insight(&cx, insert("a.jpg", 100, "m1")).unwrap();
dao.store_insight(&cx, insert("a.jpg", 200, "m2")).unwrap();
dao.store_insight(&cx, insert("a.jpg", 300, "m3")).unwrap();
dao.store_insight(insert("a.jpg", 100, "m1")).unwrap();
dao.store_insight(insert("a.jpg", 200, "m2")).unwrap();
dao.store_insight(insert("a.jpg", 300, "m3")).unwrap();
// A different path must not leak into the history.
dao.store_insight(&cx, insert("b.jpg", 250, "other"))
.unwrap();
dao.store_insight(insert("b.jpg", 250, "other")).unwrap();
let history = dao.get_insight_history(&cx, "a.jpg").unwrap();
let history = dao.get_insight_history("a.jpg").unwrap();
assert_eq!(history.len(), 3);
assert_eq!(
history.iter().map(|i| i.generated_at).collect::<Vec<_>>(),
@@ -493,20 +402,19 @@ mod tests {
#[test]
fn rate_insight_by_id_rates_only_the_targeted_version() {
let cx = opentelemetry::Context::new();
let mut dao = dao();
dao.store_insight(&cx, insert("a.jpg", 100, "m1")).unwrap();
dao.store_insight(&cx, insert("a.jpg", 200, "m2")).unwrap();
dao.store_insight(insert("a.jpg", 100, "m1")).unwrap();
dao.store_insight(insert("a.jpg", 200, "m2")).unwrap();
// History is newest-first: [200 (current), 100 (superseded)].
let history = dao.get_insight_history(&cx, "a.jpg").unwrap();
let history = dao.get_insight_history("a.jpg").unwrap();
let old_version = history.iter().find(|i| i.generated_at == 100).unwrap();
assert!(!old_version.is_current);
dao.rate_insight_by_id(&cx, old_version.id, true).unwrap();
dao.rate_insight_by_id(old_version.id, true).unwrap();
let history = dao.get_insight_history(&cx, "a.jpg").unwrap();
let history = dao.get_insight_history("a.jpg").unwrap();
let old = history.iter().find(|i| i.generated_at == 100).unwrap();
let current = history.iter().find(|i| i.generated_at == 200).unwrap();
assert_eq!(old.approved, Some(true), "targeted version is rated");
File diff suppressed because it is too large. Load diff
+11 -25
View File
@@ -43,16 +43,12 @@ pub struct InsertLocationRecord {
pub trait LocationHistoryDao: Sync + Send {
/// Store single location record
fn store_location(
&mut self,
context: &opentelemetry::Context,
location: InsertLocationRecord,
) -> Result<LocationRecord, DbError>;
fn store_location(&mut self, location: InsertLocationRecord)
-> Result<LocationRecord, DbError>;
/// Batch insert locations (Google Takeout has millions of points)
fn store_locations_batch(
&mut self,
context: &opentelemetry::Context,
locations: Vec<InsertLocationRecord>,
) -> Result<usize, DbError>;
@@ -60,7 +56,6 @@ pub trait LocationHistoryDao: Sync + Send {
/// "Where was I at photo timestamp ±N minutes?"
fn find_nearest_location(
&mut self,
context: &opentelemetry::Context,
timestamp: i64,
max_time_diff_seconds: i64,
) -> Result<Option<LocationRecord>, DbError>;
@@ -68,7 +63,6 @@ pub trait LocationHistoryDao: Sync + Send {
/// Find locations in time range
fn find_locations_in_range(
&mut self,
context: &opentelemetry::Context,
start_ts: i64,
end_ts: i64,
) -> Result<Vec<LocationRecord>, DbError>;
@@ -77,7 +71,6 @@ pub trait LocationHistoryDao: Sync + Send {
/// Uses approximate bounding box for performance
fn find_locations_near_point(
&mut self,
context: &opentelemetry::Context,
latitude: f64,
longitude: f64,
radius_km: f64,
@@ -86,14 +79,13 @@ pub trait LocationHistoryDao: Sync + Send {
/// Deduplicate: check if location exists
fn location_exists(
&mut self,
context: &opentelemetry::Context,
timestamp: i64,
latitude: f64,
longitude: f64,
) -> Result<bool, DbError>;
/// Get count of location records
fn get_location_count(&mut self, context: &opentelemetry::Context) -> Result<i64, DbError>;
fn get_location_count(&mut self) -> Result<i64, DbError>;
}
pub struct SqliteLocationHistoryDao {
@@ -205,10 +197,9 @@ struct LastInsertRowId {
impl LocationHistoryDao for SqliteLocationHistoryDao {
fn store_location(
&mut self,
context: &opentelemetry::Context,
location: InsertLocationRecord,
) -> Result<LocationRecord, DbError> {
trace_db_call(context, "insert", "store_location", |_span| {
trace_db_call("insert", "store_location", |_span| {
let mut conn = self
.connection
.lock()
@@ -279,10 +270,9 @@ impl LocationHistoryDao for SqliteLocationHistoryDao {
fn store_locations_batch(
&mut self,
context: &opentelemetry::Context,
locations: Vec<InsertLocationRecord>,
) -> Result<usize, DbError> {
trace_db_call(context, "insert", "store_locations_batch", |_span| {
trace_db_call("insert", "store_locations_batch", |_span| {
let mut conn = self
.connection
.lock()
@@ -356,11 +346,10 @@ impl LocationHistoryDao for SqliteLocationHistoryDao {
fn find_nearest_location(
&mut self,
context: &opentelemetry::Context,
timestamp: i64,
max_time_diff_seconds: i64,
) -> Result<Option<LocationRecord>, DbError> {
trace_db_call(context, "query", "find_nearest_location", |_span| {
trace_db_call("query", "find_nearest_location", |_span| {
let mut conn = self
.connection
.lock()
@@ -391,11 +380,10 @@ impl LocationHistoryDao for SqliteLocationHistoryDao {
fn find_locations_in_range(
&mut self,
context: &opentelemetry::Context,
start_ts: i64,
end_ts: i64,
) -> Result<Vec<LocationRecord>, DbError> {
trace_db_call(context, "query", "find_locations_in_range", |_span| {
trace_db_call("query", "find_locations_in_range", |_span| {
let mut conn = self
.connection
.lock()
@@ -419,12 +407,11 @@ impl LocationHistoryDao for SqliteLocationHistoryDao {
fn find_locations_near_point(
&mut self,
context: &opentelemetry::Context,
latitude: f64,
longitude: f64,
radius_km: f64,
) -> Result<Vec<LocationRecord>, DbError> {
trace_db_call(context, "query", "find_locations_near_point", |_span| {
trace_db_call("query", "find_locations_near_point", |_span| {
let mut conn = self
.connection
.lock()
@@ -474,12 +461,11 @@ impl LocationHistoryDao for SqliteLocationHistoryDao {
fn location_exists(
&mut self,
context: &opentelemetry::Context,
timestamp: i64,
latitude: f64,
longitude: f64,
) -> Result<bool, DbError> {
trace_db_call(context, "query", "location_exists", |_span| {
trace_db_call("query", "location_exists", |_span| {
let mut conn = self
.connection
.lock()
@@ -506,8 +492,8 @@ impl LocationHistoryDao for SqliteLocationHistoryDao {
.map_err(|e| DbError::log(DbErrorKind::QueryError, e))
}
fn get_location_count(&mut self, context: &opentelemetry::Context) -> Result<i64, DbError> {
trace_db_call(context, "query", "get_location_count", |_span| {
fn get_location_count(&mut self) -> Result<i64, DbError> {
trace_db_call("query", "get_location_count", |_span| {
let mut conn = self
.connection
.lock()
+244 -475
View File
File diff suppressed because it is too large. Load diff
+70 -167
View File
@@ -77,22 +77,12 @@ pub struct ImportPersona {
}
pub trait PersonaDao: Sync + Send {
fn list_personas(
&mut self,
cx: &opentelemetry::Context,
user_id: i32,
) -> Result<Vec<Persona>, DbError>;
fn list_personas(&mut self, user_id: i32) -> Result<Vec<Persona>, DbError>;
fn get_persona(
&mut self,
cx: &opentelemetry::Context,
user_id: i32,
persona_id: &str,
) -> Result<Option<Persona>, DbError>;
fn get_persona(&mut self, user_id: i32, persona_id: &str) -> Result<Option<Persona>, DbError>;
fn create_persona(
&mut self,
cx: &opentelemetry::Context,
user_id: i32,
persona_id: &str,
name: &str,
@@ -103,28 +93,17 @@ pub trait PersonaDao: Sync + Send {
fn update_persona(
&mut self,
cx: &opentelemetry::Context,
user_id: i32,
persona_id: &str,
patch: PersonaPatch,
) -> Result<Option<Persona>, DbError>;
fn delete_persona(
&mut self,
cx: &opentelemetry::Context,
user_id: i32,
persona_id: &str,
) -> Result<bool, DbError>;
fn delete_persona(&mut self, user_id: i32, persona_id: &str) -> Result<bool, DbError>;
/// Idempotent bulk import. INSERT OR IGNORE on (user_id, persona_id)
/// — re-uploading the same set is a no-op. Returns the number of rows
/// actually inserted (skipped duplicates don't count).
fn bulk_import(
&mut self,
cx: &opentelemetry::Context,
user_id: i32,
personas: &[ImportPersona],
) -> Result<usize, DbError>;
fn bulk_import(&mut self, user_id: i32, personas: &[ImportPersona]) -> Result<usize, DbError>;
// ── Persona-chat (open chat with persona) persistence ───────────
//
@@ -137,7 +116,6 @@ pub trait PersonaDao: Sync + Send {
/// another user — the two are deliberately indistinguishable to callers.
fn get_persona_chat(
&mut self,
cx: &opentelemetry::Context,
user_id: i32,
conversation_id: &str,
) -> Result<Option<PersonaChatRow>, DbError>;
@@ -145,18 +123,13 @@ pub trait PersonaDao: Sync + Send {
/// Every conversation this user has going, newest first. Backs the chat
/// list screen, so it returns the transcript blob too — the snippet is
/// derived from the tree rather than denormalized into its own column.
fn list_persona_chats(
&mut self,
cx: &opentelemetry::Context,
user_id: i32,
) -> Result<Vec<PersonaChatRow>, DbError>;
fn list_persona_chats(&mut self, user_id: i32) -> Result<Vec<PersonaChatRow>, DbError>;
/// Start a new conversation with a persona, returning its id. Several
/// conversations with the same persona are expected, so this never
/// reuses an existing row.
fn create_persona_chat(
&mut self,
cx: &opentelemetry::Context,
user_id: i32,
persona_id: &str,
created_at: i64,
@@ -169,7 +142,6 @@ pub trait PersonaDao: Sync + Send {
/// this user's.
fn update_persona_chat(
&mut self,
cx: &opentelemetry::Context,
user_id: i32,
conversation_id: &str,
messages_json: &str,
@@ -181,7 +153,6 @@ pub trait PersonaDao: Sync + Send {
/// enough of an exchange to summarize.
fn set_persona_chat_title(
&mut self,
cx: &opentelemetry::Context,
user_id: i32,
conversation_id: &str,
title: &str,
@@ -189,12 +160,7 @@ pub trait PersonaDao: Sync + Send {
/// Delete a conversation outright. Backs both the list screen's delete
/// affordance and the reset endpoint.
fn delete_persona_chat(
&mut self,
cx: &opentelemetry::Context,
user_id: i32,
conversation_id: &str,
) -> Result<(), DbError>;
fn delete_persona_chat(&mut self, user_id: i32, conversation_id: &str) -> Result<(), DbError>;
}
pub struct SqlitePersonaDao {
@@ -220,12 +186,8 @@ impl SqlitePersonaDao {
}
impl PersonaDao for SqlitePersonaDao {
fn list_personas(
&mut self,
cx: &opentelemetry::Context,
uid: i32,
) -> Result<Vec<Persona>, DbError> {
trace_db_call(cx, "query", "list_personas", |_span| {
fn list_personas(&mut self, uid: i32) -> Result<Vec<Persona>, DbError> {
trace_db_call("query", "list_personas", |_span| {
use schema::personas::dsl::*;
let mut conn = self.connection.lock().expect("PersonaDao lock");
personas
@@ -237,13 +199,8 @@ impl PersonaDao for SqlitePersonaDao {
.map_err(|e| DbError::log(DbErrorKind::QueryError, e))
}
fn get_persona(
&mut self,
cx: &opentelemetry::Context,
uid: i32,
pid: &str,
) -> Result<Option<Persona>, DbError> {
trace_db_call(cx, "query", "get_persona", |_span| {
fn get_persona(&mut self, uid: i32, pid: &str) -> Result<Option<Persona>, DbError> {
trace_db_call("query", "get_persona", |_span| {
use schema::personas::dsl::*;
let mut conn = self.connection.lock().expect("PersonaDao lock");
personas
@@ -258,7 +215,6 @@ impl PersonaDao for SqlitePersonaDao {
fn create_persona(
&mut self,
cx: &opentelemetry::Context,
uid: i32,
pid: &str,
nm: &str,
@@ -266,7 +222,7 @@ impl PersonaDao for SqlitePersonaDao {
builtin: bool,
include_all: bool,
) -> Result<Persona, DbError> {
trace_db_call(cx, "insert", "create_persona", |_span| {
trace_db_call("insert", "create_persona", |_span| {
use schema::personas::dsl::*;
let mut conn = self.connection.lock().expect("PersonaDao lock");
let now = chrono::Utc::now().timestamp_millis();
@@ -298,12 +254,11 @@ impl PersonaDao for SqlitePersonaDao {
fn update_persona(
&mut self,
cx: &opentelemetry::Context,
uid: i32,
pid: &str,
patch: PersonaPatch,
) -> Result<Option<Persona>, DbError> {
trace_db_call(cx, "update", "update_persona", |_span| {
trace_db_call("update", "update_persona", |_span| {
use schema::personas::dsl::*;
let mut conn = self.connection.lock().expect("PersonaDao lock");
let now = chrono::Utc::now().timestamp_millis();
@@ -359,13 +314,8 @@ impl PersonaDao for SqlitePersonaDao {
.map_err(|e| DbError::log(DbErrorKind::UpdateError, e))
}
fn delete_persona(
&mut self,
cx: &opentelemetry::Context,
uid: i32,
pid: &str,
) -> Result<bool, DbError> {
trace_db_call(cx, "delete", "delete_persona", |_span| {
fn delete_persona(&mut self, uid: i32, pid: &str) -> Result<bool, DbError> {
trace_db_call("delete", "delete_persona", |_span| {
let mut conn = self.connection.lock().expect("PersonaDao lock");
// One transaction so a persona and its conversations can't get
// out of step. There is no FK between the two tables, and the
@@ -392,13 +342,8 @@ impl PersonaDao for SqlitePersonaDao {
.map_err(|e| DbError::log(DbErrorKind::QueryError, e))
}
fn bulk_import(
&mut self,
cx: &opentelemetry::Context,
uid: i32,
rows: &[ImportPersona],
) -> Result<usize, DbError> {
trace_db_call(cx, "insert", "bulk_import_personas", |_span| {
fn bulk_import(&mut self, uid: i32, rows: &[ImportPersona]) -> Result<usize, DbError> {
trace_db_call("insert", "bulk_import_personas", |_span| {
let mut conn = self.connection.lock().expect("PersonaDao lock");
let now = chrono::Utc::now().timestamp_millis();
let mut inserted = 0usize;
@@ -428,13 +373,8 @@ impl PersonaDao for SqlitePersonaDao {
.map_err(|e| DbError::log(DbErrorKind::InsertError, e))
}
fn get_persona_chat(
&mut self,
cx: &opentelemetry::Context,
uid: i32,
cid: &str,
) -> Result<Option<PersonaChatRow>, DbError> {
trace_db_call(cx, "query", "get_persona_chat", |_span| {
fn get_persona_chat(&mut self, uid: i32, cid: &str) -> Result<Option<PersonaChatRow>, DbError> {
trace_db_call("query", "get_persona_chat", |_span| {
use schema::persona_chat_conversations::dsl::*;
let mut conn = self.connection.lock().expect("PersonaDao lock");
persona_chat_conversations
@@ -459,12 +399,8 @@ impl PersonaDao for SqlitePersonaDao {
.map_err(|e| DbError::log(DbErrorKind::QueryError, e))
}
fn list_persona_chats(
&mut self,
cx: &opentelemetry::Context,
uid: i32,
) -> Result<Vec<PersonaChatRow>, DbError> {
trace_db_call(cx, "query", "list_persona_chats", |_span| {
fn list_persona_chats(&mut self, uid: i32) -> Result<Vec<PersonaChatRow>, DbError> {
trace_db_call("query", "list_persona_chats", |_span| {
use schema::persona_chat_conversations::dsl::*;
let mut conn = self.connection.lock().expect("PersonaDao lock");
persona_chat_conversations
@@ -488,12 +424,11 @@ impl PersonaDao for SqlitePersonaDao {
fn create_persona_chat(
&mut self,
cx: &opentelemetry::Context,
uid: i32,
pid: &str,
created: i64,
) -> Result<String, DbError> {
trace_db_call(cx, "insert", "create_persona_chat", |_span| {
trace_db_call("insert", "create_persona_chat", |_span| {
use schema::persona_chat_conversations::dsl::*;
let mut conn = self.connection.lock().expect("PersonaDao lock");
let new_id = uuid::Uuid::new_v4().to_string();
@@ -519,14 +454,13 @@ impl PersonaDao for SqlitePersonaDao {
fn update_persona_chat(
&mut self,
cx: &opentelemetry::Context,
uid: i32,
cid: &str,
json: &str,
count: i32,
updated: i64,
) -> Result<usize, DbError> {
trace_db_call(cx, "update", "update_persona_chat", |_span| {
trace_db_call("update", "update_persona_chat", |_span| {
use schema::persona_chat_conversations::dsl::*;
let mut conn = self.connection.lock().expect("PersonaDao lock");
diesel::update(
@@ -547,12 +481,11 @@ impl PersonaDao for SqlitePersonaDao {
fn set_persona_chat_title(
&mut self,
cx: &opentelemetry::Context,
uid: i32,
cid: &str,
new_title: &str,
) -> Result<(), DbError> {
trace_db_call(cx, "update", "set_persona_chat_title", |_span| {
trace_db_call("update", "set_persona_chat_title", |_span| {
use schema::persona_chat_conversations::dsl::*;
let mut conn = self.connection.lock().expect("PersonaDao lock");
diesel::update(
@@ -568,13 +501,8 @@ impl PersonaDao for SqlitePersonaDao {
.map_err(|e| DbError::log(DbErrorKind::QueryError, e))
}
fn delete_persona_chat(
&mut self,
cx: &opentelemetry::Context,
uid: i32,
cid: &str,
) -> Result<(), DbError> {
trace_db_call(cx, "delete", "delete_persona_chat", |_span| {
fn delete_persona_chat(&mut self, uid: i32, cid: &str) -> Result<(), DbError> {
trace_db_call("delete", "delete_persona_chat", |_span| {
use schema::persona_chat_conversations::dsl::*;
let mut conn = self.connection.lock().expect("PersonaDao lock");
diesel::delete(
@@ -611,31 +539,29 @@ mod tests {
#[test]
fn create_and_list_round_trip() {
let cx = opentelemetry::Context::new();
let (mut dao, uid) = dao_with_user("alice");
// The migration seeds 3 built-ins for any existing user; alice
// was created post-migration so she starts empty.
let p = dao
.create_persona(&cx, uid, "custom-1", "Custom A", "prompt A", false, false)
.create_persona(uid, "custom-1", "Custom A", "prompt A", false, false)
.unwrap();
assert_eq!(p.persona_id, "custom-1");
assert_eq!(p.user_id, uid);
assert!(!p.is_built_in);
let list = dao.list_personas(&cx, uid).unwrap();
let list = dao.list_personas(uid).unwrap();
assert_eq!(list.len(), 1);
assert_eq!(list[0].persona_id, "custom-1");
}
#[test]
fn unique_constraint_blocks_duplicate_persona_id() {
let cx = opentelemetry::Context::new();
let (mut dao, uid) = dao_with_user("bob");
dao.create_persona(&cx, uid, "x", "X", "p", false, false)
dao.create_persona(uid, "x", "X", "p", false, false)
.unwrap();
let err = dao.create_persona(&cx, uid, "x", "X2", "p2", false, false);
let err = dao.create_persona(uid, "x", "X2", "p2", false, false);
assert!(
err.is_err(),
"second insert with same persona_id should fail"
@@ -644,7 +570,6 @@ mod tests {
#[test]
fn bulk_import_is_idempotent() {
let cx = opentelemetry::Context::new();
let (mut dao, uid) = dao_with_user("carol");
let rows = vec![
@@ -664,12 +589,12 @@ mod tests {
},
];
let first = dao.bulk_import(&cx, uid, &rows).unwrap();
let first = dao.bulk_import(uid, &rows).unwrap();
assert_eq!(first, 2);
let second = dao.bulk_import(&cx, uid, &rows).unwrap();
let second = dao.bulk_import(uid, &rows).unwrap();
assert_eq!(second, 0, "re-import should insert nothing");
assert_eq!(dao.list_personas(&cx, uid).unwrap().len(), 2);
assert_eq!(dao.list_personas(uid).unwrap().len(), 2);
}
#[test]
@@ -683,14 +608,12 @@ mod tests {
// depth — but keep this test passing so anyone who removes
// the handler guard gets a failing call site, not silent data
// corruption.
let cx = opentelemetry::Context::new();
let (mut dao, uid) = dao_with_user("eve");
dao.create_persona(&cx, uid, "default", "Default", "old", true, false)
dao.create_persona(uid, "default", "Default", "old", true, false)
.unwrap();
let updated = dao
.update_persona(
&cx,
uid,
"default",
PersonaPatch {
@@ -713,14 +636,12 @@ mod tests {
#[test]
fn update_toggles_include_all_memories() {
let cx = opentelemetry::Context::new();
let (mut dao, uid) = dao_with_user("dan");
dao.create_persona(&cx, uid, "j", "Journal", "p", true, false)
dao.create_persona(uid, "j", "Journal", "p", true, false)
.unwrap();
let updated = dao
.update_persona(
&cx,
uid,
"j",
PersonaPatch {
@@ -755,22 +676,16 @@ mod tests {
#[test]
fn persona_chat_get_returns_none_for_an_unknown_conversation() {
let cx = opentelemetry::Context::new();
let (mut dao, uid) = dao_with_user("p1");
assert!(
dao.get_persona_chat(&cx, uid, "no-such-id")
.unwrap()
.is_none()
);
assert!(dao.get_persona_chat(uid, "no-such-id").unwrap().is_none());
}
#[test]
fn persona_chat_create_then_get_round_trip() {
let cx = opentelemetry::Context::new();
let (mut dao, uid) = dao_with_user("p2");
let cid = dao.create_persona_chat(&cx, uid, "journal", 100).unwrap();
let cid = dao.create_persona_chat(uid, "journal", 100).unwrap();
let row = dao.get_persona_chat(&cx, uid, &cid).unwrap().unwrap();
let row = dao.get_persona_chat(uid, &cid).unwrap().unwrap();
assert_eq!(row.conversation_id, cid);
assert_eq!(row.persona_id, "journal");
assert_eq!(row.title, "", "unnamed until the first turn completes");
@@ -783,31 +698,27 @@ mod tests {
fn persona_chat_create_makes_a_separate_row_each_time() {
// The whole point of the conversation_id key: one persona, several
// independent threads.
let cx = opentelemetry::Context::new();
let (mut dao, uid) = dao_with_user("p3");
let first = dao.create_persona_chat(&cx, uid, "journal", 100).unwrap();
let second = dao.create_persona_chat(&cx, uid, "journal", 200).unwrap();
let first = dao.create_persona_chat(uid, "journal", 100).unwrap();
let second = dao.create_persona_chat(uid, "journal", 200).unwrap();
assert_ne!(first, second);
assert_eq!(dao.list_persona_chats(&cx, uid).unwrap().len(), 2);
assert_eq!(dao.list_persona_chats(uid).unwrap().len(), 2);
}
#[test]
fn persona_chat_update_replaces_the_transcript() {
let cx = opentelemetry::Context::new();
let (mut dao, uid) = dao_with_user("p4");
let cid = dao.create_persona_chat(&cx, uid, "journal", 100).unwrap();
let cid = dao.create_persona_chat(uid, "journal", 100).unwrap();
let rows = dao
.update_persona_chat(&cx, uid, &cid, "first", 1, 150)
.unwrap();
let rows = dao.update_persona_chat(uid, &cid, "first", 1, 150).unwrap();
assert_eq!(rows, 1);
let rows = dao
.update_persona_chat(&cx, uid, &cid, "second", 2, 200)
.update_persona_chat(uid, &cid, "second", 2, 200)
.unwrap();
assert_eq!(rows, 1);
let row = dao.get_persona_chat(&cx, uid, &cid).unwrap().unwrap();
let row = dao.get_persona_chat(uid, &cid).unwrap().unwrap();
assert_eq!(row.messages_json, "second");
assert_eq!(row.turn_count, 2);
assert_eq!(row.updated_at, 200);
@@ -818,36 +729,32 @@ mod tests {
fn persona_chat_update_reports_zero_rows_for_an_unknown_conversation() {
// The turn loop treats 0 as "the conversation went away mid-turn"
// rather than silently succeeding.
let cx = opentelemetry::Context::new();
let (mut dao, uid) = dao_with_user("p5");
let rows = dao
.update_persona_chat(&cx, uid, "no-such-id", "x", 1, 1)
.update_persona_chat(uid, "no-such-id", "x", 1, 1)
.unwrap();
assert_eq!(rows, 0);
}
#[test]
fn persona_chat_set_title_names_the_conversation() {
let cx = opentelemetry::Context::new();
let (mut dao, uid) = dao_with_user("p6");
let cid = dao.create_persona_chat(&cx, uid, "journal", 100).unwrap();
let cid = dao.create_persona_chat(uid, "journal", 100).unwrap();
dao.set_persona_chat_title(&cx, uid, &cid, "June recap")
.unwrap();
let row = dao.get_persona_chat(&cx, uid, &cid).unwrap().unwrap();
dao.set_persona_chat_title(uid, &cid, "June recap").unwrap();
let row = dao.get_persona_chat(uid, &cid).unwrap().unwrap();
assert_eq!(row.title, "June recap");
}
#[test]
fn persona_chat_list_orders_newest_first() {
let cx = opentelemetry::Context::new();
let (mut dao, uid) = dao_with_user("p7");
let older = dao.create_persona_chat(&cx, uid, "journal", 100).unwrap();
let newer = dao.create_persona_chat(&cx, uid, "coach", 300).unwrap();
let middle = dao.create_persona_chat(&cx, uid, "default", 200).unwrap();
let older = dao.create_persona_chat(uid, "journal", 100).unwrap();
let newer = dao.create_persona_chat(uid, "coach", 300).unwrap();
let middle = dao.create_persona_chat(uid, "default", 200).unwrap();
let ids: Vec<String> = dao
.list_persona_chats(&cx, uid)
.list_persona_chats(uid)
.unwrap()
.into_iter()
.map(|r| r.conversation_id)
@@ -861,20 +768,18 @@ mod tests {
#[test]
fn persona_chat_list_is_empty_for_a_user_who_has_never_chatted() {
let cx = opentelemetry::Context::new();
let (mut dao, uid) = dao_with_user("p8");
assert!(dao.list_persona_chats(&cx, uid).unwrap().is_empty());
assert!(dao.list_persona_chats(uid).unwrap().is_empty());
}
#[test]
fn persona_chat_delete_removes_the_conversation() {
let cx = opentelemetry::Context::new();
let (mut dao, uid) = dao_with_user("p9");
let cid = dao.create_persona_chat(&cx, uid, "journal", 100).unwrap();
let cid = dao.create_persona_chat(uid, "journal", 100).unwrap();
dao.delete_persona_chat(&cx, uid, &cid).unwrap();
assert!(dao.get_persona_chat(&cx, uid, &cid).unwrap().is_none());
assert!(dao.list_persona_chats(&cx, uid).unwrap().is_empty());
dao.delete_persona_chat(uid, &cid).unwrap();
assert!(dao.get_persona_chat(uid, &cid).unwrap().is_none());
assert!(dao.list_persona_chats(uid).unwrap().is_empty());
}
#[test]
@@ -882,18 +787,17 @@ mod tests {
// No FK between the tables, and the chat list hides conversations
// whose persona is gone — so an orphan row would be an invisible,
// unreachable transcript sitting in the database forever.
let cx = opentelemetry::Context::new();
let (mut dao, uid) = dao_with_user("p-cascade");
dao.create_persona(&cx, uid, "custom-1", "Custom", "prompt", false, false)
dao.create_persona(uid, "custom-1", "Custom", "prompt", false, false)
.unwrap();
dao.create_persona_chat(&cx, uid, "custom-1", 100).unwrap();
dao.create_persona_chat(&cx, uid, "custom-1", 200).unwrap();
let survivor = dao.create_persona_chat(&cx, uid, "other", 300).unwrap();
dao.create_persona_chat(uid, "custom-1", 100).unwrap();
dao.create_persona_chat(uid, "custom-1", 200).unwrap();
let survivor = dao.create_persona_chat(uid, "other", 300).unwrap();
assert!(dao.delete_persona(&cx, uid, "custom-1").unwrap());
assert!(dao.delete_persona(uid, "custom-1").unwrap());
let remaining: Vec<String> = dao
.list_persona_chats(&cx, uid)
.list_persona_chats(uid)
.unwrap()
.into_iter()
.map(|r| r.conversation_id)
@@ -909,27 +813,26 @@ mod tests {
fn persona_chat_reads_and_writes_are_scoped_to_the_owner() {
// A conversation_id is a bearer token for someone's transcript, so
// holding one must not grant another user access to it.
let cx = opentelemetry::Context::new();
let (mut dao, uid1) = dao_with_user("owner");
let uid2 = second_user(&dao, "intruder");
let cid = dao.create_persona_chat(&cx, uid1, "journal", 100).unwrap();
dao.update_persona_chat(&cx, uid1, &cid, "private", 1, 100)
let cid = dao.create_persona_chat(uid1, "journal", 100).unwrap();
dao.update_persona_chat(uid1, &cid, "private", 1, 100)
.unwrap();
assert!(
dao.get_persona_chat(&cx, uid2, &cid).unwrap().is_none(),
dao.get_persona_chat(uid2, &cid).unwrap().is_none(),
"another user cannot read it"
);
assert_eq!(
dao.update_persona_chat(&cx, uid2, &cid, "tampered", 9, 999)
dao.update_persona_chat(uid2, &cid, "tampered", 9, 999)
.unwrap(),
0,
"another user cannot write it"
);
dao.delete_persona_chat(&cx, uid2, &cid).unwrap();
dao.delete_persona_chat(uid2, &cid).unwrap();
let row = dao.get_persona_chat(&cx, uid1, &cid).unwrap().unwrap();
let row = dao.get_persona_chat(uid1, &cid).unwrap().unwrap();
assert_eq!(row.messages_json, "private", "and cannot delete it");
assert!(dao.list_persona_chats(&cx, uid2).unwrap().is_empty());
assert!(dao.list_persona_chats(uid2).unwrap().is_empty());
}
}
+45 -87
View File
@@ -15,16 +15,11 @@ pub trait PrecomputedReelDao: Sync + Send {
/// Insert a precomputed reel row. Returns the new row's id.
/// Written by the nightly agentic job (Section D).
#[allow(dead_code)]
fn record_reel(
&mut self,
context: &opentelemetry::Context,
row: &InsertablePrecomputedReel,
) -> Result<i32, DbError>;
fn record_reel(&mut self, row: &InsertablePrecomputedReel) -> Result<i32, DbError>;
/// Find the latest precomputed reel for the given (span, library_key).
fn latest_for(
&mut self,
context: &opentelemetry::Context,
span: &str,
library_key: &str,
) -> Result<Option<PrecomputedReel>, DbError>;
@@ -35,7 +30,6 @@ pub trait PrecomputedReelDao: Sync + Send {
/// back to `latest_for` (avoids a second query path).
fn exists_fresh(
&mut self,
context: &opentelemetry::Context,
span: &str,
library_key: &str,
render_version: i32,
@@ -48,7 +42,6 @@ pub trait PrecomputedReelDao: Sync + Send {
#[allow(dead_code)]
fn prune_superseded(
&mut self,
context: &opentelemetry::Context,
span: &str,
library_key: &str,
keep: usize,
@@ -57,25 +50,20 @@ pub trait PrecomputedReelDao: Sync + Send {
/// Every cache_key currently in the ledger. Used by the on-disk cache sweep
/// to protect files a ledger row still points at.
#[allow(dead_code)]
fn all_cache_keys(&mut self, context: &opentelemetry::Context) -> Result<Vec<String>, DbError>;
fn all_cache_keys(&mut self) -> Result<Vec<String>, DbError>;
/// Every reel row for `span` whose `library_key` is one of
/// `library_keys`. Used by the true-instant date drain (ticket #118)
/// to find the day reels a re-dated photo invalidates.
fn list_for_span(
&mut self,
context: &opentelemetry::Context,
span: &str,
library_keys: &[String],
) -> Result<Vec<PrecomputedReel>, DbError>;
/// Delete the given reel rows (stale-marking: the nightly pre-gen
/// finds no current row and regenerates). Returns rows deleted.
fn delete_ids(
&mut self,
context: &opentelemetry::Context,
ids: &[i32],
) -> Result<usize, DbError>;
fn delete_ids(&mut self, ids: &[i32]) -> Result<usize, DbError>;
}
pub struct SqlitePrecomputedReelDao {
@@ -102,12 +90,8 @@ impl SqlitePrecomputedReelDao {
}
impl PrecomputedReelDao for SqlitePrecomputedReelDao {
fn record_reel(
&mut self,
context: &opentelemetry::Context,
row: &InsertablePrecomputedReel,
) -> Result<i32, DbError> {
trace_db_call(context, "insert", "record_reel", |_span| {
fn record_reel(&mut self, row: &InsertablePrecomputedReel) -> Result<i32, DbError> {
trace_db_call("insert", "record_reel", |_span| {
use schema::precomputed_reels::dsl;
let mut connection = self
@@ -131,11 +115,10 @@ impl PrecomputedReelDao for SqlitePrecomputedReelDao {
fn latest_for(
&mut self,
context: &opentelemetry::Context,
span: &str,
library_key: &str,
) -> Result<Option<PrecomputedReel>, DbError> {
trace_db_call(context, "query", "latest_for", |_span| {
trace_db_call("query", "latest_for", |_span| {
use schema::precomputed_reels::dsl;
let mut connection = self
@@ -156,13 +139,12 @@ impl PrecomputedReelDao for SqlitePrecomputedReelDao {
fn exists_fresh(
&mut self,
context: &opentelemetry::Context,
span: &str,
library_key: &str,
render_version: i32,
min_generated_at: i64,
) -> Result<bool, DbError> {
trace_db_call(context, "query", "exists_fresh", |_span| {
trace_db_call("query", "exists_fresh", |_span| {
use schema::precomputed_reels::dsl;
let mut connection = self
@@ -186,12 +168,11 @@ impl PrecomputedReelDao for SqlitePrecomputedReelDao {
fn prune_superseded(
&mut self,
context: &opentelemetry::Context,
span: &str,
library_key: &str,
keep: usize,
) -> Result<Vec<PrecomputedReel>, DbError> {
trace_db_call(context, "delete", "prune_superseded", |_span| {
trace_db_call("delete", "prune_superseded", |_span| {
use schema::precomputed_reels::dsl;
let mut connection = self
@@ -221,8 +202,8 @@ impl PrecomputedReelDao for SqlitePrecomputedReelDao {
.map_err(|e| DbError::log(DbErrorKind::UpdateError, e))
}
fn all_cache_keys(&mut self, context: &opentelemetry::Context) -> Result<Vec<String>, DbError> {
trace_db_call(context, "query", "all_cache_keys", |_span| {
fn all_cache_keys(&mut self) -> Result<Vec<String>, DbError> {
trace_db_call("query", "all_cache_keys", |_span| {
use schema::precomputed_reels::dsl;
let mut connection = self
@@ -240,11 +221,10 @@ impl PrecomputedReelDao for SqlitePrecomputedReelDao {
fn list_for_span(
&mut self,
context: &opentelemetry::Context,
span: &str,
library_keys: &[String],
) -> Result<Vec<PrecomputedReel>, DbError> {
trace_db_call(context, "query", "list_for_span", |_span| {
trace_db_call("query", "list_for_span", |_span| {
use schema::precomputed_reels::dsl;
let mut connection = self
@@ -261,12 +241,8 @@ impl PrecomputedReelDao for SqlitePrecomputedReelDao {
.map_err(|e| DbError::log(DbErrorKind::QueryError, e))
}
fn delete_ids(
&mut self,
context: &opentelemetry::Context,
ids: &[i32],
) -> Result<usize, DbError> {
trace_db_call(context, "delete", "delete_ids", |_span| {
fn delete_ids(&mut self, ids: &[i32]) -> Result<usize, DbError> {
trace_db_call("delete", "delete_ids", |_span| {
use schema::precomputed_reels::dsl;
let mut connection = self
@@ -298,10 +274,6 @@ mod tests {
SqlitePrecomputedReelDao::from_connection(Arc::new(Mutex::new(conn)))
}
fn ctx() -> opentelemetry::Context {
opentelemetry::Context::new()
}
fn sample_row() -> InsertablePrecomputedReel {
InsertablePrecomputedReel {
span: "day".to_string(),
@@ -320,28 +292,25 @@ mod tests {
#[test]
fn record_reel_inserts_and_returns_id() {
let mut dao = setup_dao();
let ctx = ctx();
let row = sample_row();
let id = dao.record_reel(&ctx, &row).unwrap();
let id = dao.record_reel(&row).unwrap();
assert!(id > 0, "should return a positive id");
}
#[test]
fn record_reel_returns_increasing_ids() {
let mut dao = setup_dao();
let ctx = ctx();
let row = sample_row();
let id1 = dao.record_reel(&ctx, &row).unwrap();
let id2 = dao.record_reel(&ctx, &row).unwrap();
let id1 = dao.record_reel(&row).unwrap();
let id2 = dao.record_reel(&row).unwrap();
assert!(id2 > id1, "each insert should get a higher id");
}
#[test]
fn latest_for_returns_latest() {
let mut dao = setup_dao();
let ctx = ctx();
let row1 = InsertablePrecomputedReel {
generated_at: 1_000_000,
@@ -352,17 +321,16 @@ mod tests {
..sample_row()
};
dao.record_reel(&ctx, &row1).unwrap();
dao.record_reel(&ctx, &row2).unwrap();
dao.record_reel(&row1).unwrap();
dao.record_reel(&row2).unwrap();
let latest = dao.latest_for(&ctx, "day", "1").unwrap().unwrap();
let latest = dao.latest_for("day", "1").unwrap().unwrap();
assert_eq!(latest.generated_at, 2_000_000);
}
#[test]
fn latest_for_scoped_by_span_and_library() {
let mut dao = setup_dao();
let ctx = ctx();
let day_row = InsertablePrecomputedReel {
span: "day".to_string(),
@@ -377,94 +345,85 @@ mod tests {
..sample_row()
};
dao.record_reel(&ctx, &day_row).unwrap();
dao.record_reel(&ctx, &week_row).unwrap();
dao.record_reel(&day_row).unwrap();
dao.record_reel(&week_row).unwrap();
let day_latest = dao.latest_for(&ctx, "day", "1").unwrap().unwrap();
let day_latest = dao.latest_for("day", "1").unwrap().unwrap();
assert_eq!(day_latest.span, "day");
let week_latest = dao.latest_for(&ctx, "week", "1").unwrap().unwrap();
let week_latest = dao.latest_for("week", "1").unwrap().unwrap();
assert_eq!(week_latest.span, "week");
// Different library returns None
let missing = dao.latest_for(&ctx, "day", "99").unwrap();
let missing = dao.latest_for("day", "99").unwrap();
assert!(missing.is_none());
}
#[test]
fn latest_for_returns_none_when_no_rows() {
let mut dao = setup_dao();
let ctx = ctx();
let result = dao.latest_for(&ctx, "day", "1").unwrap();
let result = dao.latest_for("day", "1").unwrap();
assert!(result.is_none());
}
#[test]
fn exists_fresh_returns_true_when_present() {
let mut dao = setup_dao();
let ctx = ctx();
dao.record_reel(&ctx, &sample_row()).unwrap();
dao.record_reel(&sample_row()).unwrap();
let exists = dao.exists_fresh(&ctx, "day", "1", 1, 900_000).unwrap();
let exists = dao.exists_fresh("day", "1", 1, 900_000).unwrap();
assert!(exists, "should find the row we just inserted");
}
#[test]
fn exists_fresh_returns_false_when_missing() {
let mut dao = setup_dao();
let ctx = ctx();
let exists = dao.exists_fresh(&ctx, "day", "1", 1, 900_000).unwrap();
let exists = dao.exists_fresh("day", "1", 1, 900_000).unwrap();
assert!(!exists, "should not find anything in empty table");
}
#[test]
fn exists_fresh_respects_min_generated_at() {
let mut dao = setup_dao();
let ctx = ctx();
dao.record_reel(&ctx, &sample_row()).unwrap();
dao.record_reel(&sample_row()).unwrap();
// Below the threshold — should exist
let exists = dao.exists_fresh(&ctx, "day", "1", 1, 500_000).unwrap();
let exists = dao.exists_fresh("day", "1", 1, 500_000).unwrap();
assert!(exists);
// Above the threshold — should not exist
let exists = dao.exists_fresh(&ctx, "day", "1", 1, 2_000_000).unwrap();
let exists = dao.exists_fresh("day", "1", 1, 2_000_000).unwrap();
assert!(!exists);
}
#[test]
fn exists_fresh_respects_render_version() {
let mut dao = setup_dao();
let ctx = ctx();
let row_v1 = InsertablePrecomputedReel {
render_version: 1,
..sample_row()
};
dao.record_reel(&ctx, &row_v1).unwrap();
dao.record_reel(&row_v1).unwrap();
assert!(dao.exists_fresh(&ctx, "day", "1", 1, 900_000).unwrap());
assert!(!dao.exists_fresh(&ctx, "day", "1", 2, 900_000).unwrap());
assert!(dao.exists_fresh("day", "1", 1, 900_000).unwrap());
assert!(!dao.exists_fresh("day", "1", 2, 900_000).unwrap());
}
#[test]
fn prune_superseded_keeps_newest_and_returns_deleted() {
let mut dao = setup_dao();
let ctx = ctx();
// Three day/lib1 reels at increasing timestamps, plus an unrelated one.
for (i, key) in ["k1", "k2", "k3"].iter().enumerate() {
dao.record_reel(
&ctx,
&InsertablePrecomputedReel {
cache_key: key.to_string(),
generated_at: 1_000_000 + i as i64 * 1000,
..sample_row()
},
)
dao.record_reel(&InsertablePrecomputedReel {
cache_key: key.to_string(),
generated_at: 1_000_000 + i as i64 * 1000,
..sample_row()
})
.unwrap();
}
let other = InsertablePrecomputedReel {
@@ -472,15 +431,15 @@ mod tests {
cache_key: "other".to_string(),
..sample_row()
};
dao.record_reel(&ctx, &other).unwrap();
dao.record_reel(&other).unwrap();
// Keep the newest 2 of (day, "1"); k1 (oldest) is superseded.
let deleted = dao.prune_superseded(&ctx, "day", "1", 2).unwrap();
let deleted = dao.prune_superseded("day", "1", 2).unwrap();
assert_eq!(deleted.len(), 1);
assert_eq!(deleted[0].cache_key, "k1");
// The newest 2 survive; the other-library row is untouched.
let keys = dao.all_cache_keys(&ctx).unwrap();
let keys = dao.all_cache_keys().unwrap();
assert_eq!(keys.len(), 3);
assert!(keys.contains(&"k2".to_string()));
assert!(keys.contains(&"k3".to_string()));
@@ -491,10 +450,9 @@ mod tests {
#[test]
fn prune_superseded_noop_when_within_keep() {
let mut dao = setup_dao();
let ctx = ctx();
dao.record_reel(&ctx, &sample_row()).unwrap();
let deleted = dao.prune_superseded(&ctx, "day", "1", 2).unwrap();
dao.record_reel(&sample_row()).unwrap();
let deleted = dao.prune_superseded("day", "1", 2).unwrap();
assert!(deleted.is_empty());
assert_eq!(dao.all_cache_keys(&ctx).unwrap().len(), 1);
assert_eq!(dao.all_cache_keys().unwrap().len(), 1);
}
}
+33 -81
View File
@@ -10,16 +10,10 @@ use crate::database::{DbError, DbErrorKind, connect};
use crate::otel::trace_db_call;
pub trait PreviewDao: Sync + Send {
fn insert_preview(
&mut self,
context: &opentelemetry::Context,
file_path_val: &str,
status_val: &str,
) -> Result<(), DbError>;
fn insert_preview(&mut self, file_path_val: &str, status_val: &str) -> Result<(), DbError>;
fn update_status(
&mut self,
context: &opentelemetry::Context,
file_path_val: &str,
status_val: &str,
duration: Option<f32>,
@@ -27,23 +21,14 @@ pub trait PreviewDao: Sync + Send {
error: Option<&str>,
) -> Result<(), DbError>;
fn get_preview(
&mut self,
context: &opentelemetry::Context,
file_path_val: &str,
) -> Result<Option<VideoPreviewClip>, DbError>;
fn get_preview(&mut self, file_path_val: &str) -> Result<Option<VideoPreviewClip>, DbError>;
fn get_previews_batch(
&mut self,
context: &opentelemetry::Context,
file_paths: &[String],
) -> Result<Vec<VideoPreviewClip>, DbError>;
fn get_by_status(
&mut self,
context: &opentelemetry::Context,
status_val: &str,
) -> Result<Vec<VideoPreviewClip>, DbError>;
fn get_by_status(&mut self, status_val: &str) -> Result<Vec<VideoPreviewClip>, DbError>;
}
pub struct SqlitePreviewDao {
@@ -72,13 +57,8 @@ impl SqlitePreviewDao {
}
impl PreviewDao for SqlitePreviewDao {
fn insert_preview(
&mut self,
context: &opentelemetry::Context,
file_path_val: &str,
status_val: &str,
) -> Result<(), DbError> {
trace_db_call(context, "insert", "insert_preview", |_span| {
fn insert_preview(&mut self, file_path_val: &str, status_val: &str) -> Result<(), DbError> {
trace_db_call("insert", "insert_preview", |_span| {
use crate::database::schema::video_preview_clips::dsl::*;
let mut connection = self.connection.lock().expect("Unable to get PreviewDao");
@@ -101,14 +81,13 @@ impl PreviewDao for SqlitePreviewDao {
fn update_status(
&mut self,
context: &opentelemetry::Context,
file_path_val: &str,
status_val: &str,
duration: Option<f32>,
size: Option<i32>,
error: Option<&str>,
) -> Result<(), DbError> {
trace_db_call(context, "update", "update_preview_status", |_span| {
trace_db_call("update", "update_preview_status", |_span| {
use crate::database::schema::video_preview_clips::dsl::*;
let mut connection = self.connection.lock().expect("Unable to get PreviewDao");
@@ -129,12 +108,8 @@ impl PreviewDao for SqlitePreviewDao {
.map_err(|e| DbError::log(DbErrorKind::UpdateError, e))
}
fn get_preview(
&mut self,
context: &opentelemetry::Context,
file_path_val: &str,
) -> Result<Option<VideoPreviewClip>, DbError> {
trace_db_call(context, "query", "get_preview", |_span| {
fn get_preview(&mut self, file_path_val: &str) -> Result<Option<VideoPreviewClip>, DbError> {
trace_db_call("query", "get_preview", |_span| {
use crate::database::schema::video_preview_clips::dsl::*;
let mut connection = self.connection.lock().expect("Unable to get PreviewDao");
@@ -153,10 +128,9 @@ impl PreviewDao for SqlitePreviewDao {
fn get_previews_batch(
&mut self,
context: &opentelemetry::Context,
file_paths: &[String],
) -> Result<Vec<VideoPreviewClip>, DbError> {
trace_db_call(context, "query", "get_previews_batch", |_span| {
trace_db_call("query", "get_previews_batch", |_span| {
use crate::database::schema::video_preview_clips::dsl::*;
if file_paths.is_empty() {
@@ -173,12 +147,8 @@ impl PreviewDao for SqlitePreviewDao {
.map_err(|e| DbError::log(DbErrorKind::QueryError, e))
}
fn get_by_status(
&mut self,
context: &opentelemetry::Context,
status_val: &str,
) -> Result<Vec<VideoPreviewClip>, DbError> {
trace_db_call(context, "query", "get_previews_by_status", |_span| {
fn get_by_status(&mut self, status_val: &str) -> Result<Vec<VideoPreviewClip>, DbError> {
trace_db_call("query", "get_previews_by_status", |_span| {
use crate::database::schema::video_preview_clips::dsl::*;
let mut connection = self.connection.lock().expect("Unable to get PreviewDao");
@@ -201,19 +171,13 @@ mod tests {
SqlitePreviewDao::from_connection(in_memory_db_connection())
}
fn ctx() -> opentelemetry::Context {
opentelemetry::Context::new()
}
#[test]
fn test_insert_and_get_preview() {
let mut dao = setup_dao();
let ctx = ctx();
dao.insert_preview(&ctx, "photos/video.mp4", "pending")
.unwrap();
dao.insert_preview("photos/video.mp4", "pending").unwrap();
let result = dao.get_preview(&ctx, "photos/video.mp4").unwrap();
let result = dao.get_preview("photos/video.mp4").unwrap();
assert!(result.is_some());
let clip = result.unwrap();
assert_eq!(clip.file_path, "photos/video.mp4");
@@ -226,28 +190,23 @@ mod tests {
#[test]
fn test_insert_duplicate_ignored() {
let mut dao = setup_dao();
let ctx = ctx();
dao.insert_preview(&ctx, "photos/video.mp4", "pending")
.unwrap();
dao.insert_preview("photos/video.mp4", "pending").unwrap();
// Second insert with same path should not error (INSERT OR IGNORE)
dao.insert_preview(&ctx, "photos/video.mp4", "processing")
dao.insert_preview("photos/video.mp4", "processing")
.unwrap();
// Status should remain "pending" from the first insert
let clip = dao.get_preview(&ctx, "photos/video.mp4").unwrap().unwrap();
let clip = dao.get_preview("photos/video.mp4").unwrap().unwrap();
assert_eq!(clip.status, "pending");
}
#[test]
fn test_update_status_to_complete() {
let mut dao = setup_dao();
let ctx = ctx();
dao.insert_preview(&ctx, "photos/video.mp4", "pending")
.unwrap();
dao.insert_preview("photos/video.mp4", "pending").unwrap();
dao.update_status(
&ctx,
"photos/video.mp4",
"complete",
Some(9.5),
@@ -256,7 +215,7 @@ mod tests {
)
.unwrap();
let clip = dao.get_preview(&ctx, "photos/video.mp4").unwrap().unwrap();
let clip = dao.get_preview("photos/video.mp4").unwrap().unwrap();
assert_eq!(clip.status, "complete");
assert_eq!(clip.duration_seconds, Some(9.5));
assert_eq!(clip.file_size_bytes, Some(1024000));
@@ -266,12 +225,9 @@ mod tests {
#[test]
fn test_update_status_to_failed() {
let mut dao = setup_dao();
let ctx = ctx();
dao.insert_preview(&ctx, "photos/video.mp4", "pending")
.unwrap();
dao.insert_preview("photos/video.mp4", "pending").unwrap();
dao.update_status(
&ctx,
"photos/video.mp4",
"failed",
None,
@@ -280,7 +236,7 @@ mod tests {
)
.unwrap();
let clip = dao.get_preview(&ctx, "photos/video.mp4").unwrap().unwrap();
let clip = dao.get_preview("photos/video.mp4").unwrap().unwrap();
assert_eq!(clip.status, "failed");
assert_eq!(
clip.error_message.as_deref(),
@@ -291,24 +247,22 @@ mod tests {
#[test]
fn test_get_preview_not_found() {
let mut dao = setup_dao();
let ctx = ctx();
let result = dao.get_preview(&ctx, "nonexistent/path.mp4").unwrap();
let result = dao.get_preview("nonexistent/path.mp4").unwrap();
assert!(result.is_none());
}
#[test]
fn test_get_previews_batch() {
let mut dao = setup_dao();
let ctx = ctx();
dao.insert_preview(&ctx, "a/one.mp4", "complete").unwrap();
dao.insert_preview(&ctx, "b/two.mp4", "pending").unwrap();
dao.insert_preview(&ctx, "c/three.mp4", "failed").unwrap();
dao.insert_preview("a/one.mp4", "complete").unwrap();
dao.insert_preview("b/two.mp4", "pending").unwrap();
dao.insert_preview("c/three.mp4", "failed").unwrap();
// Query only two of the three
let paths = vec!["a/one.mp4".to_string(), "c/three.mp4".to_string()];
let results = dao.get_previews_batch(&ctx, &paths).unwrap();
let results = dao.get_previews_batch(&paths).unwrap();
assert_eq!(results.len(), 2);
let statuses: Vec<&str> = results.iter().map(|c| c.status.as_str()).collect();
@@ -319,30 +273,28 @@ mod tests {
#[test]
fn test_get_previews_batch_empty_input() {
let mut dao = setup_dao();
let ctx = ctx();
let results = dao.get_previews_batch(&ctx, &[]).unwrap();
let results = dao.get_previews_batch(&[]).unwrap();
assert!(results.is_empty());
}
#[test]
fn test_get_by_status() {
let mut dao = setup_dao();
let ctx = ctx();
dao.insert_preview(&ctx, "a.mp4", "pending").unwrap();
dao.insert_preview(&ctx, "b.mp4", "complete").unwrap();
dao.insert_preview(&ctx, "c.mp4", "pending").unwrap();
dao.insert_preview(&ctx, "d.mp4", "failed").unwrap();
dao.insert_preview("a.mp4", "pending").unwrap();
dao.insert_preview("b.mp4", "complete").unwrap();
dao.insert_preview("c.mp4", "pending").unwrap();
dao.insert_preview("d.mp4", "failed").unwrap();
let pending = dao.get_by_status(&ctx, "pending").unwrap();
let pending = dao.get_by_status("pending").unwrap();
assert_eq!(pending.len(), 2);
let complete = dao.get_by_status(&ctx, "complete").unwrap();
let complete = dao.get_by_status("complete").unwrap();
assert_eq!(complete.len(), 1);
assert_eq!(complete[0].file_path, "b.mp4");
let processing = dao.get_by_status(&ctx, "processing").unwrap();
let processing = dao.get_by_status("processing").unwrap();
assert!(processing.is_empty());
}
}
+15 -43
View File
@@ -33,23 +33,15 @@ pub struct InsertSearchRecord {
pub trait SearchHistoryDao: Sync + Send {
/// Store search with embedding
fn store_search(
&mut self,
context: &opentelemetry::Context,
search: InsertSearchRecord,
) -> Result<SearchRecord, DbError>;
fn store_search(&mut self, search: InsertSearchRecord) -> Result<SearchRecord, DbError>;
/// Batch insert searches
fn store_searches_batch(
&mut self,
context: &opentelemetry::Context,
searches: Vec<InsertSearchRecord>,
) -> Result<usize, DbError>;
fn store_searches_batch(&mut self, searches: Vec<InsertSearchRecord>)
-> Result<usize, DbError>;
/// Find searches in time range (for temporal context)
fn find_searches_in_range(
&mut self,
context: &opentelemetry::Context,
start_ts: i64,
end_ts: i64,
) -> Result<Vec<SearchRecord>, DbError>;
@@ -57,7 +49,6 @@ pub trait SearchHistoryDao: Sync + Send {
/// Find semantically similar searches (PRIMARY - embeddings shine here)
fn find_similar_searches(
&mut self,
context: &opentelemetry::Context,
query_embedding: &[f32],
limit: usize,
) -> Result<Vec<SearchRecord>, DbError>;
@@ -65,7 +56,6 @@ pub trait SearchHistoryDao: Sync + Send {
/// Hybrid: Time window + semantic ranking
fn find_relevant_searches_hybrid(
&mut self,
context: &opentelemetry::Context,
center_timestamp: i64,
time_window_days: i64,
query_embedding: Option<&[f32]>,
@@ -73,15 +63,10 @@ pub trait SearchHistoryDao: Sync + Send {
) -> Result<Vec<SearchRecord>, DbError>;
/// Deduplication check
fn search_exists(
&mut self,
context: &opentelemetry::Context,
timestamp: i64,
query: &str,
) -> Result<bool, DbError>;
fn search_exists(&mut self, timestamp: i64, query: &str) -> Result<bool, DbError>;
/// Get count of search records
fn get_search_count(&mut self, context: &opentelemetry::Context) -> Result<i64, DbError>;
fn get_search_count(&mut self) -> Result<i64, DbError>;
}
pub struct SqliteSearchHistoryDao {
@@ -177,12 +162,8 @@ struct LastInsertRowId {
}
impl SearchHistoryDao for SqliteSearchHistoryDao {
fn store_search(
&mut self,
context: &opentelemetry::Context,
search: InsertSearchRecord,
) -> Result<SearchRecord, DbError> {
trace_db_call(context, "insert", "store_search", |_span| {
fn store_search(&mut self, search: InsertSearchRecord) -> Result<SearchRecord, DbError> {
trace_db_call("insert", "store_search", |_span| {
let mut conn = self
.connection
.lock()
@@ -233,10 +214,9 @@ impl SearchHistoryDao for SqliteSearchHistoryDao {
fn store_searches_batch(
&mut self,
context: &opentelemetry::Context,
searches: Vec<InsertSearchRecord>,
) -> Result<usize, DbError> {
trace_db_call(context, "insert", "store_searches_batch", |_span| {
trace_db_call("insert", "store_searches_batch", |_span| {
let mut conn = self
.connection
.lock()
@@ -289,11 +269,10 @@ impl SearchHistoryDao for SqliteSearchHistoryDao {
fn find_searches_in_range(
&mut self,
context: &opentelemetry::Context,
start_ts: i64,
end_ts: i64,
) -> Result<Vec<SearchRecord>, DbError> {
trace_db_call(context, "query", "find_searches_in_range", |_span| {
trace_db_call("query", "find_searches_in_range", |_span| {
let mut conn = self
.connection
.lock()
@@ -316,11 +295,10 @@ impl SearchHistoryDao for SqliteSearchHistoryDao {
fn find_similar_searches(
&mut self,
context: &opentelemetry::Context,
query_embedding: &[f32],
limit: usize,
) -> Result<Vec<SearchRecord>, DbError> {
trace_db_call(context, "query", "find_similar_searches", |_span| {
trace_db_call("query", "find_similar_searches", |_span| {
let mut conn = self
.connection
.lock()
@@ -379,13 +357,12 @@ impl SearchHistoryDao for SqliteSearchHistoryDao {
fn find_relevant_searches_hybrid(
&mut self,
context: &opentelemetry::Context,
center_timestamp: i64,
time_window_days: i64,
query_embedding: Option<&[f32]>,
limit: usize,
) -> Result<Vec<SearchRecord>, DbError> {
trace_db_call(context, "query", "find_relevant_searches_hybrid", |_span| {
trace_db_call("query", "find_relevant_searches_hybrid", |_span| {
let window_seconds = time_window_days * 86400;
let start_ts = center_timestamp - window_seconds;
let end_ts = center_timestamp + window_seconds;
@@ -465,13 +442,8 @@ impl SearchHistoryDao for SqliteSearchHistoryDao {
.map_err(|e| DbError::log(DbErrorKind::QueryError, e))
}
fn search_exists(
&mut self,
context: &opentelemetry::Context,
timestamp: i64,
query: &str,
) -> Result<bool, DbError> {
trace_db_call(context, "query", "search_exists", |_span| {
fn search_exists(&mut self, timestamp: i64, query: &str) -> Result<bool, DbError> {
trace_db_call("query", "search_exists", |_span| {
let mut conn = self
.connection
.lock()
@@ -496,8 +468,8 @@ impl SearchHistoryDao for SqliteSearchHistoryDao {
.map_err(|e| DbError::log(DbErrorKind::QueryError, e))
}
fn get_search_count(&mut self, context: &opentelemetry::Context) -> Result<i64, DbError> {
trace_db_call(context, "query", "get_search_count", |_span| {
fn get_search_count(&mut self) -> Result<i64, DbError> {
trace_db_call("query", "get_search_count", |_span| {
let mut conn = self
.connection
.lock()
+14 -32
View File
@@ -13,18 +13,11 @@ use crate::otel::trace_db_call;
/// pre-generation scheduler (Section D) to pick up user preferences.
pub trait UserAiPrefsDao: Sync + Send {
/// Read the single row; `None` when it hasn't been populated yet.
fn get_prefs(
&mut self,
context: &opentelemetry::Context,
) -> Result<Option<UserAiPrefs>, DbError>;
fn get_prefs(&mut self) -> Result<Option<UserAiPrefs>, DbError>;
/// Upsert the single row (id is always 1).
#[allow(dead_code)]
fn upsert_prefs(
&mut self,
context: &opentelemetry::Context,
prefs: &UpsertUserAiPrefs,
) -> Result<(), DbError>;
fn upsert_prefs(&mut self, prefs: &UpsertUserAiPrefs) -> Result<(), DbError>;
}
pub struct SqliteUserAiPrefsDao {
@@ -51,11 +44,8 @@ impl SqliteUserAiPrefsDao {
}
impl UserAiPrefsDao for SqliteUserAiPrefsDao {
fn get_prefs(
&mut self,
context: &opentelemetry::Context,
) -> Result<Option<UserAiPrefs>, DbError> {
trace_db_call(context, "query", "get_prefs", |_span| {
fn get_prefs(&mut self) -> Result<Option<UserAiPrefs>, DbError> {
trace_db_call("query", "get_prefs", |_span| {
use schema::user_ai_prefs::dsl;
let mut connection = self
@@ -71,12 +61,8 @@ impl UserAiPrefsDao for SqliteUserAiPrefsDao {
.map_err(|e| DbError::log(DbErrorKind::QueryError, e))
}
fn upsert_prefs(
&mut self,
context: &opentelemetry::Context,
prefs: &UpsertUserAiPrefs,
) -> Result<(), DbError> {
trace_db_call(context, "upsert", "upsert_prefs", |_span| {
fn upsert_prefs(&mut self, prefs: &UpsertUserAiPrefs) -> Result<(), DbError> {
trace_db_call("upsert", "upsert_prefs", |_span| {
use schema::user_ai_prefs::dsl;
let mut connection = self
@@ -131,14 +117,10 @@ mod tests {
SqliteUserAiPrefsDao::from_connection(Arc::new(Mutex::new(conn)))
}
fn ctx() -> opentelemetry::Context {
opentelemetry::Context::new()
}
#[test]
fn get_prefs_returns_none_when_empty() {
let mut dao = setup_dao();
let result = dao.get_prefs(&ctx()).unwrap();
let result = dao.get_prefs().unwrap();
assert!(result.is_none());
}
@@ -158,9 +140,9 @@ mod tests {
reel_span: Some("week".to_string()),
reel_max_segments: Some(24),
};
dao.upsert_prefs(&ctx(), &prefs).unwrap();
dao.upsert_prefs(&prefs).unwrap();
let row = dao.get_prefs(&ctx()).unwrap().unwrap();
let row = dao.get_prefs().unwrap().unwrap();
assert_eq!(row.id, 1);
assert_eq!(row.voice, Some("grandma".to_string()));
assert_eq!(row.tz_offset_minutes, Some(-480));
@@ -192,7 +174,7 @@ mod tests {
reel_span: Some("week".to_string()),
reel_max_segments: Some(24),
};
dao.upsert_prefs(&ctx(), &prefs1).unwrap();
dao.upsert_prefs(&prefs1).unwrap();
let prefs2 = UpsertUserAiPrefs {
voice: Some("dad".to_string()),
@@ -206,9 +188,9 @@ mod tests {
reel_span: None,
reel_max_segments: None,
};
dao.upsert_prefs(&ctx(), &prefs2).unwrap();
dao.upsert_prefs(&prefs2).unwrap();
let row = dao.get_prefs(&ctx()).unwrap().unwrap();
let row = dao.get_prefs().unwrap().unwrap();
assert_eq!(row.voice, Some("dad".to_string()));
assert_eq!(row.tz_offset_minutes, Some(-300));
assert!(row.library.is_none());
@@ -238,9 +220,9 @@ mod tests {
reel_span: Some("month".to_string()),
reel_max_segments: None,
};
dao.upsert_prefs(&ctx(), &prefs).unwrap();
dao.upsert_prefs(&prefs).unwrap();
let row = dao.get_prefs(&ctx()).unwrap().unwrap();
let row = dao.get_prefs().unwrap().unwrap();
assert_eq!(row.tz_offset_minutes, Some(-480));
assert!(row.voice.is_none());
assert!(row.library.is_none());