runtara_core/persistence/
postgres.rs

1// Copyright (C) 2025 SyncMyOrders Sp. z o.o.
2// SPDX-License-Identifier: AGPL-3.0-or-later
3//! Persistence operations for runtara-core.
4//!
5//! Provides all durable storage access functions for instances, checkpoints, events, and signals.
6
7#![allow(dead_code)] // Fields and functions used in tests and by handlers
8
9use chrono::{DateTime, Utc};
10use sqlx::PgPool;
11use uuid::Uuid;
12
13use crate::error::CoreError;
14
15/// PostgreSQL-backed persistence implementation.
16#[derive(Clone)]
17pub struct PostgresPersistence {
18    pool: PgPool,
19}
20
21impl PostgresPersistence {
22    /// Create a new Postgres-backed persistence implementation.
23    pub fn new(pool: PgPool) -> Self {
24        Self { pool }
25    }
26}
27
28// ============================================================================
29// Record Types
30// ============================================================================
31
32use super::{
33    CheckpointRecord, CustomSignalRecord, EventRecord, EventSortOrder, InstanceRecord,
34    ListEventsFilter, ListStepSummariesFilter, Persistence, SignalRecord, StepStatus,
35    StepSummaryRecord, WakeEntry,
36};
37
38// ============================================================================
39// Instance Operations
40// ============================================================================
41
42/// Create a new instance record.
43pub async fn create_instance(
44    pool: &PgPool,
45    instance_id: &str,
46    tenant_id: &str,
47) -> Result<(), CoreError> {
48    sqlx::query(
49        r#"
50        INSERT INTO instances (instance_id, tenant_id, definition_version, status, created_at)
51        VALUES ($1, $2, 1, 'pending'::instance_status, NOW())
52        "#,
53    )
54    .bind(instance_id)
55    .bind(tenant_id)
56    .execute(pool)
57    .await?;
58
59    Ok(())
60}
61
62/// UUID used for self-registered instances (no image/definition).
63/// This is a well-known UUID that indicates the instance registered itself.
64pub const SELF_REGISTERED_DEFINITION_ID: Uuid = Uuid::from_u128(0);
65
66/// Create a self-registered instance record.
67/// Used when an instance registers itself without being started via the management API.
68/// Get an instance by ID.
69pub async fn get_instance(
70    pool: &PgPool,
71    instance_id: &str,
72) -> Result<Option<InstanceRecord>, CoreError> {
73    let record = sqlx::query_as::<_, InstanceRecord>(
74        r#"
75        SELECT instance_id, tenant_id, definition_version,
76               status::text as status, checkpoint_id, attempt, max_attempts,
77               created_at, started_at, finished_at, input, output, error, sleep_until
78        FROM instances
79        WHERE instance_id = $1
80        "#,
81    )
82    .bind(instance_id)
83    .fetch_optional(pool)
84    .await?;
85
86    Ok(record)
87}
88
89/// Update instance status.
90pub async fn update_instance_status(
91    pool: &PgPool,
92    instance_id: &str,
93    status: &str,
94    started_at: Option<DateTime<Utc>>,
95) -> Result<(), CoreError> {
96    let result = if let Some(started) = started_at {
97        sqlx::query(
98            r#"
99            UPDATE instances
100            SET status = $2::instance_status, started_at = $3
101            WHERE instance_id = $1
102            "#,
103        )
104        .bind(instance_id)
105        .bind(status)
106        .bind(started)
107        .execute(pool)
108        .await?
109    } else {
110        sqlx::query(
111            r#"
112            UPDATE instances
113            SET status = $2::instance_status
114            WHERE instance_id = $1
115            "#,
116        )
117        .bind(instance_id)
118        .bind(status)
119        .execute(pool)
120        .await?
121    };
122
123    if result.rows_affected() == 0 {
124        return Err(CoreError::InstanceNotFound {
125            instance_id: instance_id.to_string(),
126        });
127    }
128
129    Ok(())
130}
131
132/// Update instance's current checkpoint ID.
133pub async fn update_instance_checkpoint(
134    pool: &PgPool,
135    instance_id: &str,
136    checkpoint_id: &str,
137) -> Result<(), CoreError> {
138    let result = sqlx::query(
139        r#"
140        UPDATE instances
141        SET checkpoint_id = $2
142        WHERE instance_id = $1
143        "#,
144    )
145    .bind(instance_id)
146    .bind(checkpoint_id)
147    .execute(pool)
148    .await?;
149
150    if result.rows_affected() == 0 {
151        return Err(CoreError::InstanceNotFound {
152            instance_id: instance_id.to_string(),
153        });
154    }
155
156    Ok(())
157}
158
159/// Mark instance as completed (success or failure).
160pub async fn complete_instance(
161    pool: &PgPool,
162    instance_id: &str,
163    output: Option<&[u8]>,
164    error: Option<&str>,
165) -> Result<(), CoreError> {
166    let status = if error.is_some() {
167        "failed"
168    } else {
169        "completed"
170    };
171
172    let result = sqlx::query(
173        r#"
174        UPDATE instances
175        SET status = $2::instance_status,
176            finished_at = NOW(),
177            output = $3,
178            error = $4
179        WHERE instance_id = $1
180        "#,
181    )
182    .bind(instance_id)
183    .bind(status)
184    .bind(output)
185    .bind(error)
186    .execute(pool)
187    .await?;
188
189    if result.rows_affected() == 0 {
190        return Err(CoreError::InstanceNotFound {
191            instance_id: instance_id.to_string(),
192        });
193    }
194
195    Ok(())
196}
197
198/// Complete an instance with extended fields (stderr, checkpoint).
199///
200/// This is the full version that supports all fields needed by Environment.
201pub async fn complete_instance_extended(
202    pool: &PgPool,
203    instance_id: &str,
204    status: &str,
205    output: Option<&[u8]>,
206    error: Option<&str>,
207    stderr: Option<&str>,
208    checkpoint_id: Option<&str>,
209) -> Result<(), CoreError> {
210    let result = sqlx::query(
211        r#"
212        UPDATE instances
213        SET status = $2::instance_status,
214            output = $3,
215            error = $4,
216            stderr = COALESCE($5, stderr),
217            checkpoint_id = COALESCE($6, checkpoint_id),
218            finished_at = CASE
219                WHEN $2 IN ('completed', 'failed', 'cancelled', 'suspended') THEN NOW()
220                ELSE finished_at
221            END
222        WHERE instance_id = $1
223        "#,
224    )
225    .bind(instance_id)
226    .bind(status)
227    .bind(output)
228    .bind(error)
229    .bind(stderr)
230    .bind(checkpoint_id)
231    .execute(pool)
232    .await?;
233
234    if result.rows_affected() == 0 {
235        return Err(CoreError::InstanceNotFound {
236            instance_id: instance_id.to_string(),
237        });
238    }
239
240    Ok(())
241}
242
243/// Complete an instance only if its current status is 'running'.
244///
245/// This prevents race conditions where both Core (via SDK) and Environment
246/// (via container monitor) try to complete the same instance.
247/// Returns true if the update was applied, false if skipped.
248pub async fn complete_instance_if_running(
249    pool: &PgPool,
250    instance_id: &str,
251    status: &str,
252    output: Option<&[u8]>,
253    error: Option<&str>,
254    stderr: Option<&str>,
255    checkpoint_id: Option<&str>,
256) -> Result<bool, CoreError> {
257    let result = sqlx::query(
258        r#"
259        UPDATE instances
260        SET status = $2::instance_status,
261            output = $3,
262            error = $4,
263            stderr = COALESCE($5, stderr),
264            checkpoint_id = COALESCE($6, checkpoint_id),
265            finished_at = CASE
266                WHEN $2 IN ('completed', 'failed', 'cancelled', 'suspended') THEN NOW()
267                ELSE finished_at
268            END
269        WHERE instance_id = $1 AND status = 'running'
270        "#,
271    )
272    .bind(instance_id)
273    .bind(status)
274    .bind(output)
275    .bind(error)
276    .bind(stderr)
277    .bind(checkpoint_id)
278    .execute(pool)
279    .await?;
280
281    Ok(result.rows_affected() > 0)
282}
283
284/// Complete an instance with termination tracking metadata.
285///
286/// This version includes `termination_reason` and `exit_code` for unambiguous
287/// identification of how/why the instance terminated.
288#[allow(clippy::too_many_arguments)]
289pub async fn complete_instance_with_termination(
290    pool: &PgPool,
291    instance_id: &str,
292    status: &str,
293    termination_reason: Option<&str>,
294    exit_code: Option<i32>,
295    output: Option<&[u8]>,
296    error: Option<&str>,
297    stderr: Option<&str>,
298    checkpoint_id: Option<&str>,
299) -> Result<(), CoreError> {
300    sqlx::query(
301        r#"
302        UPDATE instances
303        SET status = $2::instance_status,
304            termination_reason = COALESCE($3::termination_reason, termination_reason),
305            exit_code = COALESCE($4, exit_code),
306            output = $5,
307            error = $6,
308            stderr = COALESCE($7, stderr),
309            checkpoint_id = COALESCE($8, checkpoint_id),
310            finished_at = CASE
311                WHEN $2 IN ('completed', 'failed', 'cancelled', 'suspended') THEN NOW()
312                ELSE finished_at
313            END
314        WHERE instance_id = $1
315        "#,
316    )
317    .bind(instance_id)
318    .bind(status)
319    .bind(termination_reason)
320    .bind(exit_code)
321    .bind(output)
322    .bind(error)
323    .bind(stderr)
324    .bind(checkpoint_id)
325    .execute(pool)
326    .await?;
327
328    Ok(())
329}
330
331/// Complete an instance with termination tracking, only if status is 'running'.
332///
333/// Returns true if the update was applied, false if skipped (instance already
334/// has a terminal status).
335#[allow(clippy::too_many_arguments)]
336pub async fn complete_instance_with_termination_if_running(
337    pool: &PgPool,
338    instance_id: &str,
339    status: &str,
340    termination_reason: Option<&str>,
341    exit_code: Option<i32>,
342    output: Option<&[u8]>,
343    error: Option<&str>,
344    stderr: Option<&str>,
345    checkpoint_id: Option<&str>,
346) -> Result<bool, CoreError> {
347    let result = sqlx::query(
348        r#"
349        UPDATE instances
350        SET status = $2::instance_status,
351            termination_reason = COALESCE($3::termination_reason, termination_reason),
352            exit_code = COALESCE($4, exit_code),
353            output = $5,
354            error = $6,
355            stderr = COALESCE($7, stderr),
356            checkpoint_id = COALESCE($8, checkpoint_id),
357            finished_at = CASE
358                WHEN $2 IN ('completed', 'failed', 'cancelled', 'suspended') THEN NOW()
359                ELSE finished_at
360            END
361        WHERE instance_id = $1 AND status = 'running'
362        "#,
363    )
364    .bind(instance_id)
365    .bind(status)
366    .bind(termination_reason)
367    .bind(exit_code)
368    .bind(output)
369    .bind(error)
370    .bind(stderr)
371    .bind(checkpoint_id)
372    .execute(pool)
373    .await?;
374
375    Ok(result.rows_affected() > 0)
376}
377
378/// Update execution metrics for an instance.
379///
380/// Stores cgroup-collected resource usage metrics (memory, CPU) after container execution.
381/// Only updates if metrics are not already set (first writer wins).
382pub async fn update_instance_metrics(
383    pool: &PgPool,
384    instance_id: &str,
385    memory_peak_bytes: Option<u64>,
386    cpu_usage_usec: Option<u64>,
387) -> Result<(), CoreError> {
388    sqlx::query(
389        r#"
390        UPDATE instances
391        SET memory_peak_bytes = COALESCE(memory_peak_bytes, $2),
392            cpu_usage_usec = COALESCE(cpu_usage_usec, $3)
393        WHERE instance_id = $1
394        "#,
395    )
396    .bind(instance_id)
397    .bind(memory_peak_bytes.map(|v| v as i64))
398    .bind(cpu_usage_usec.map(|v| v as i64))
399    .execute(pool)
400    .await?;
401
402    Ok(())
403}
404
405/// Update instance stderr (raw container stderr output).
406///
407/// Stores stderr from container execution for debugging/logging purposes.
408/// Only updates if stderr is not already set (first writer wins).
409pub async fn update_instance_stderr(
410    pool: &PgPool,
411    instance_id: &str,
412    stderr: &str,
413) -> Result<(), CoreError> {
414    sqlx::query(
415        r#"
416        UPDATE instances
417        SET stderr = COALESCE(stderr, $2)
418        WHERE instance_id = $1
419        "#,
420    )
421    .bind(instance_id)
422    .bind(stderr)
423    .execute(pool)
424    .await?;
425
426    Ok(())
427}
428
429/// Store input data for an instance.
430pub async fn store_instance_input(
431    pool: &PgPool,
432    instance_id: &str,
433    input: &[u8],
434) -> Result<(), CoreError> {
435    sqlx::query(
436        r#"
437        UPDATE instances
438        SET input = $2
439        WHERE instance_id = $1
440        "#,
441    )
442    .bind(instance_id)
443    .bind(input)
444    .execute(pool)
445    .await?;
446
447    Ok(())
448}
449
450// ============================================================================
451// Checkpoint Operations
452// ============================================================================
453
454/// Save a checkpoint (append-only).
455/// Uses ON CONFLICT to handle duplicate checkpoint IDs gracefully.
456pub async fn save_checkpoint(
457    pool: &PgPool,
458    instance_id: &str,
459    checkpoint_id: &str,
460    state: &[u8],
461) -> Result<(), CoreError> {
462    sqlx::query(
463        r#"
464        INSERT INTO checkpoints (instance_id, checkpoint_id, state, created_at)
465        VALUES ($1, $2, $3, NOW())
466        ON CONFLICT (instance_id, checkpoint_id) DO UPDATE
467        SET state = EXCLUDED.state, created_at = NOW()
468        "#,
469    )
470    .bind(instance_id)
471    .bind(checkpoint_id)
472    .bind(state)
473    .execute(pool)
474    .await
475    .map_err(|e| CoreError::CheckpointSaveFailed {
476        instance_id: instance_id.to_string(),
477        reason: e.to_string(),
478    })?;
479
480    Ok(())
481}
482
483/// Load a specific checkpoint by ID.
484pub async fn load_checkpoint(
485    pool: &PgPool,
486    instance_id: &str,
487    checkpoint_id: &str,
488) -> Result<Option<CheckpointRecord>, CoreError> {
489    let record = sqlx::query_as::<_, CheckpointRecord>(
490        r#"
491        SELECT id, instance_id, checkpoint_id, state, created_at
492        FROM checkpoints
493        WHERE instance_id = $1 AND checkpoint_id = $2
494        "#,
495    )
496    .bind(instance_id)
497    .bind(checkpoint_id)
498    .fetch_optional(pool)
499    .await?;
500
501    Ok(record)
502}
503
504/// Load the latest checkpoint for an instance.
505pub async fn load_latest_checkpoint(
506    pool: &PgPool,
507    instance_id: &str,
508) -> Result<Option<CheckpointRecord>, CoreError> {
509    let record = sqlx::query_as::<_, CheckpointRecord>(
510        r#"
511        SELECT id, instance_id, checkpoint_id, state, created_at
512        FROM checkpoints
513        WHERE instance_id = $1
514        ORDER BY created_at DESC
515        LIMIT 1
516        "#,
517    )
518    .bind(instance_id)
519    .fetch_optional(pool)
520    .await?;
521
522    Ok(record)
523}
524
525/// List checkpoints for an instance with filtering and pagination.
526pub async fn list_checkpoints(
527    pool: &PgPool,
528    instance_id: &str,
529    checkpoint_id_filter: Option<&str>,
530    limit: i64,
531    offset: i64,
532    created_after: Option<DateTime<Utc>>,
533    created_before: Option<DateTime<Utc>>,
534) -> Result<Vec<CheckpointRecord>, CoreError> {
535    let records = sqlx::query_as::<_, CheckpointRecord>(
536        r#"
537        SELECT id, instance_id, checkpoint_id, state, created_at
538        FROM checkpoints
539        WHERE instance_id = $1
540          AND ($2::TEXT IS NULL OR checkpoint_id = $2)
541          AND ($3::TIMESTAMPTZ IS NULL OR created_at >= $3)
542          AND ($4::TIMESTAMPTZ IS NULL OR created_at < $4)
543        ORDER BY created_at DESC
544        LIMIT $5 OFFSET $6
545        "#,
546    )
547    .bind(instance_id)
548    .bind(checkpoint_id_filter)
549    .bind(created_after)
550    .bind(created_before)
551    .bind(limit)
552    .bind(offset)
553    .fetch_all(pool)
554    .await?;
555
556    Ok(records)
557}
558
559/// Retry attempt record from the database.
560/// These are stored in the checkpoints table with is_retry_attempt = true.
561#[derive(Debug, Clone, sqlx::FromRow)]
562pub struct RetryAttemptRecord {
563    /// Database primary key.
564    pub id: i64,
565    /// Instance this retry attempt belongs to.
566    pub instance_id: String,
567    /// Base checkpoint identifier (the durable function's cache key).
568    pub checkpoint_id: String,
569    /// Retry attempt number (1-indexed).
570    pub attempt_number: i32,
571    /// Error message from this attempt.
572    pub error_message: Option<String>,
573    /// When the retry attempt was recorded.
574    pub created_at: DateTime<Utc>,
575}
576
577/// Save a retry attempt record for audit trail.
578/// Retry attempts are stored in the checkpoints table with a unique checkpoint_id.
579pub async fn save_retry_attempt(
580    pool: &PgPool,
581    instance_id: &str,
582    checkpoint_id: &str,
583    attempt_number: i32,
584    error_message: Option<&str>,
585) -> Result<(), CoreError> {
586    // Create a unique checkpoint_id for this retry attempt
587    let retry_checkpoint_id = format!("{}::retry::{}", checkpoint_id, attempt_number);
588
589    sqlx::query(
590        r#"
591        INSERT INTO checkpoints (instance_id, checkpoint_id, state, is_retry_attempt, attempt_number, error_message, created_at)
592        VALUES ($1, $2, '', true, $3, $4, NOW())
593        ON CONFLICT (instance_id, checkpoint_id) DO UPDATE
594        SET attempt_number = EXCLUDED.attempt_number,
595            error_message = EXCLUDED.error_message,
596            created_at = NOW()
597        "#,
598    )
599    .bind(instance_id)
600    .bind(&retry_checkpoint_id)
601    .bind(attempt_number)
602    .bind(error_message)
603    .execute(pool)
604    .await
605    .map_err(|e| CoreError::CheckpointSaveFailed {
606        instance_id: instance_id.to_string(),
607        reason: e.to_string(),
608    })?;
609
610    Ok(())
611}
612
613/// Load retry history for a checkpoint (for debugging/audit).
614/// Returns all retry attempts for the given base checkpoint_id.
615pub async fn load_retry_history(
616    pool: &PgPool,
617    instance_id: &str,
618    checkpoint_id: &str,
619) -> Result<Vec<RetryAttemptRecord>, CoreError> {
620    let pattern = format!("{}::retry::%", checkpoint_id);
621
622    let records = sqlx::query_as::<_, RetryAttemptRecord>(
623        r#"
624        SELECT id, instance_id, checkpoint_id, attempt_number, error_message, created_at
625        FROM checkpoints
626        WHERE instance_id = $1
627          AND checkpoint_id LIKE $2
628          AND is_retry_attempt = true
629        ORDER BY attempt_number ASC
630        "#,
631    )
632    .bind(instance_id)
633    .bind(&pattern)
634    .fetch_all(pool)
635    .await?;
636
637    Ok(records)
638}
639
640/// Count checkpoints for an instance with filtering.
641pub async fn count_checkpoints(
642    pool: &PgPool,
643    instance_id: &str,
644    checkpoint_id_filter: Option<&str>,
645    created_after: Option<DateTime<Utc>>,
646    created_before: Option<DateTime<Utc>>,
647) -> Result<i64, CoreError> {
648    let count: (i64,) = sqlx::query_as(
649        r#"
650        SELECT COUNT(*)
651        FROM checkpoints
652        WHERE instance_id = $1
653          AND ($2::TEXT IS NULL OR checkpoint_id = $2)
654          AND ($3::TIMESTAMPTZ IS NULL OR created_at >= $3)
655          AND ($4::TIMESTAMPTZ IS NULL OR created_at < $4)
656        "#,
657    )
658    .bind(instance_id)
659    .bind(checkpoint_id_filter)
660    .bind(created_after)
661    .bind(created_before)
662    .fetch_one(pool)
663    .await?;
664
665    Ok(count.0)
666}
667
668// ============================================================================
669// Wake Queue Operations
670// ============================================================================
671
672/// Schedule a wake for an instance.
673/// Uses ON CONFLICT to replace existing wake entry for the same instance.
674pub async fn schedule_wake(
675    pool: &PgPool,
676    instance_id: &str,
677    checkpoint_id: &str,
678    wake_at: DateTime<Utc>,
679) -> Result<(), CoreError> {
680    sqlx::query(
681        r#"
682        INSERT INTO wake_queue (instance_id, checkpoint_id, wake_at, created_at)
683        VALUES ($1, $2, $3, NOW())
684        ON CONFLICT (instance_id) DO UPDATE
685        SET checkpoint_id = EXCLUDED.checkpoint_id,
686            wake_at = EXCLUDED.wake_at,
687            created_at = NOW()
688        "#,
689    )
690    .bind(instance_id)
691    .bind(checkpoint_id)
692    .bind(wake_at)
693    .execute(pool)
694    .await?;
695
696    Ok(())
697}
698
699/// Get a wake entry for an instance.
700pub async fn get_wake_entry(
701    pool: &PgPool,
702    instance_id: &str,
703) -> Result<Option<WakeEntry>, CoreError> {
704    let record = sqlx::query_as::<_, WakeEntry>(
705        r#"
706        SELECT id, instance_id, checkpoint_id, wake_at, created_at
707        FROM wake_queue
708        WHERE instance_id = $1
709        "#,
710    )
711    .bind(instance_id)
712    .fetch_optional(pool)
713    .await?;
714
715    Ok(record)
716}
717
718/// Clear wake entry for an instance.
719pub async fn clear_wake(pool: &PgPool, instance_id: &str) -> Result<(), CoreError> {
720    sqlx::query(
721        r#"
722        DELETE FROM wake_queue
723        WHERE instance_id = $1
724        "#,
725    )
726    .bind(instance_id)
727    .execute(pool)
728    .await?;
729
730    Ok(())
731}
732
733// ============================================================================
734// Event Operations
735// ============================================================================
736
737/// Insert an instance event.
738pub async fn insert_event(pool: &PgPool, event: &EventRecord) -> Result<(), CoreError> {
739    sqlx::query(
740        r#"
741        INSERT INTO instance_events (instance_id, event_type, checkpoint_id, payload, created_at, subtype)
742        VALUES ($1, $2::instance_event_type, $3, $4, $5, $6)
743        "#,
744    )
745    .bind(&event.instance_id)
746    .bind(&event.event_type)
747    .bind(&event.checkpoint_id)
748    .bind(&event.payload)
749    .bind(event.created_at)
750    .bind(&event.subtype)
751    .execute(pool)
752    .await?;
753
754    Ok(())
755}
756
757/// List events for an instance with filtering and pagination.
758///
759/// Supports filtering by event_type, subtype, time range, full-text search
760/// in the JSON payload, and scope hierarchy filtering. Sort order can be
761/// configured via filter.sort_order (defaults to DESC - newest first).
762pub async fn list_events(
763    pool: &PgPool,
764    instance_id: &str,
765    filter: &ListEventsFilter,
766    limit: i64,
767    offset: i64,
768) -> Result<Vec<EventRecord>, CoreError> {
769    // For PostgreSQL, we use convert_from to search text within BYTEA payload
770    // The payload is expected to be valid UTF-8 JSON when subtype is set
771    // Scope filtering uses JSONB operators on the payload for efficient querying
772
773    // Determine sort order - ASC or DESC based on filter
774    let order_direction = match filter.sort_order {
775        EventSortOrder::Asc => "ASC",
776        EventSortOrder::Desc => "DESC",
777    };
778
779    // Build query with dynamic ORDER BY
780    // Note: ORDER BY direction cannot be parameterized, so we use format!
781    // The direction is from a trusted enum, so this is safe from injection
782    let query = format!(
783        r#"
784        SELECT id, instance_id, event_type::text as event_type, checkpoint_id, payload, created_at, subtype
785        FROM instance_events
786        WHERE instance_id = $1
787          AND ($2::TEXT IS NULL OR event_type::text = $2)
788          AND ($3::TEXT IS NULL OR subtype = $3)
789          AND ($4::TIMESTAMPTZ IS NULL OR created_at >= $4)
790          AND ($5::TIMESTAMPTZ IS NULL OR created_at < $5)
791          AND ($6::TEXT IS NULL OR (
792              payload IS NOT NULL
793              AND convert_from(payload, 'UTF8') ILIKE '%' || $6 || '%'
794          ))
795          AND ($7::TEXT IS NULL OR (
796              payload IS NOT NULL
797              AND convert_from(payload, 'UTF8')::jsonb->>'scope_id' = $7
798          ))
799          AND ($8::TEXT IS NULL OR (
800              payload IS NOT NULL
801              AND convert_from(payload, 'UTF8')::jsonb->>'parent_scope_id' = $8
802          ))
803          AND (NOT $9 OR (
804              payload IS NULL
805              OR convert_from(payload, 'UTF8')::jsonb->>'parent_scope_id' IS NULL
806          ))
807        ORDER BY created_at {}, id {}
808        LIMIT $10 OFFSET $11
809        "#,
810        order_direction, order_direction
811    );
812
813    let records = sqlx::query_as::<_, EventRecord>(&query)
814        .bind(instance_id)
815        .bind(&filter.event_type)
816        .bind(&filter.subtype)
817        .bind(filter.created_after)
818        .bind(filter.created_before)
819        .bind(&filter.payload_contains)
820        .bind(&filter.scope_id)
821        .bind(&filter.parent_scope_id)
822        .bind(filter.root_scopes_only)
823        .bind(limit)
824        .bind(offset)
825        .fetch_all(pool)
826        .await?;
827
828    Ok(records)
829}
830
831/// Count events for an instance with filtering (including scope hierarchy filtering).
832pub async fn count_events(
833    pool: &PgPool,
834    instance_id: &str,
835    filter: &ListEventsFilter,
836) -> Result<i64, CoreError> {
837    let count: (i64,) = sqlx::query_as(
838        r#"
839        SELECT COUNT(*)
840        FROM instance_events
841        WHERE instance_id = $1
842          AND ($2::TEXT IS NULL OR event_type::text = $2)
843          AND ($3::TEXT IS NULL OR subtype = $3)
844          AND ($4::TIMESTAMPTZ IS NULL OR created_at >= $4)
845          AND ($5::TIMESTAMPTZ IS NULL OR created_at < $5)
846          AND ($6::TEXT IS NULL OR (
847              payload IS NOT NULL
848              AND convert_from(payload, 'UTF8') ILIKE '%' || $6 || '%'
849          ))
850          AND ($7::TEXT IS NULL OR (
851              payload IS NOT NULL
852              AND convert_from(payload, 'UTF8')::jsonb->>'scope_id' = $7
853          ))
854          AND ($8::TEXT IS NULL OR (
855              payload IS NOT NULL
856              AND convert_from(payload, 'UTF8')::jsonb->>'parent_scope_id' = $8
857          ))
858          AND (NOT $9 OR (
859              payload IS NULL
860              OR convert_from(payload, 'UTF8')::jsonb->>'parent_scope_id' IS NULL
861          ))
862        "#,
863    )
864    .bind(instance_id)
865    .bind(&filter.event_type)
866    .bind(&filter.subtype)
867    .bind(filter.created_after)
868    .bind(filter.created_before)
869    .bind(&filter.payload_contains)
870    .bind(&filter.scope_id)
871    .bind(&filter.parent_scope_id)
872    .bind(filter.root_scopes_only)
873    .fetch_one(pool)
874    .await?;
875
876    Ok(count.0)
877}
878
879// ============================================================================
880// Step Summary Operations (paired step_debug_start/end events)
881// ============================================================================
882
883/// List step summaries for an instance, pairing step_debug_start and step_debug_end events.
884///
885/// Uses a CTE to join start and end events by step_id, computing status and duration.
886/// Supports filtering by status, step_type, scope hierarchy, and sorting.
887pub async fn list_step_summaries(
888    pool: &PgPool,
889    instance_id: &str,
890    filter: &ListStepSummariesFilter,
891    limit: i64,
892    offset: i64,
893) -> Result<Vec<StepSummaryRecord>, CoreError> {
894    // Determine sort order
895    let order_direction = match filter.sort_order {
896        EventSortOrder::Asc => "ASC",
897        EventSortOrder::Desc => "DESC",
898    };
899
900    // Convert status filter to string for SQL CASE matching
901    let status_filter: Option<&str> = filter.status.map(|s| match s {
902        StepStatus::Running => "running",
903        StepStatus::Completed => "completed",
904        StepStatus::Failed => "failed",
905    });
906
907    // Build query with dynamic ORDER BY
908    let query = format!(
909        r#"
910        WITH start_events AS (
911            SELECT
912                id,
913                convert_from(payload, 'UTF8')::jsonb->>'step_id' as step_id,
914                convert_from(payload, 'UTF8')::jsonb->>'step_name' as step_name,
915                convert_from(payload, 'UTF8')::jsonb->>'step_type' as step_type,
916                convert_from(payload, 'UTF8')::jsonb->>'scope_id' as scope_id,
917                convert_from(payload, 'UTF8')::jsonb->>'parent_scope_id' as parent_scope_id,
918                convert_from(payload, 'UTF8')::jsonb->'inputs' as inputs,
919                created_at
920            FROM instance_events
921            WHERE instance_id = $1 AND subtype = 'step_debug_start'
922        ),
923        end_events AS (
924            SELECT
925                convert_from(payload, 'UTF8')::jsonb->>'step_id' as step_id,
926                convert_from(payload, 'UTF8')::jsonb->>'scope_id' as scope_id,
927                convert_from(payload, 'UTF8')::jsonb->'outputs' as outputs,
928                convert_from(payload, 'UTF8')::jsonb->'error' as error,
929                created_at
930            FROM instance_events
931            WHERE instance_id = $1 AND subtype = 'step_debug_end'
932        ),
933        paired AS (
934            SELECT
935                s.step_id,
936                s.step_name,
937                s.step_type,
938                s.scope_id,
939                s.parent_scope_id,
940                s.inputs,
941                s.created_at as started_at,
942                e.created_at as completed_at,
943                e.outputs,
944                e.error,
945                CASE
946                    WHEN e.step_id IS NULL THEN 'running'
947                    WHEN e.error IS NOT NULL AND e.error::text != 'null' THEN 'failed'
948                    ELSE 'completed'
949                END as status,
950                CASE
951                    WHEN e.created_at IS NOT NULL
952                    THEN EXTRACT(MILLISECONDS FROM (e.created_at - s.created_at))::bigint
953                    ELSE NULL
954                END as duration_ms,
955                s.id as sort_id
956            FROM start_events s
957            LEFT JOIN end_events e ON s.step_id = e.step_id AND COALESCE(s.scope_id, '') = COALESCE(e.scope_id, '')
958        )
959        SELECT
960            step_id, step_name, step_type, scope_id, parent_scope_id,
961            inputs, started_at, completed_at, outputs, error, status, duration_ms
962        FROM paired
963        WHERE ($2::TEXT IS NULL OR status = $2)
964          AND ($3::TEXT IS NULL OR step_type = $3)
965          AND ($4::TEXT IS NULL OR scope_id = $4)
966          AND ($5::TEXT IS NULL OR parent_scope_id = $5)
967          AND (NOT $6 OR parent_scope_id IS NULL)
968        ORDER BY sort_id {}
969        LIMIT $7 OFFSET $8
970        "#,
971        order_direction
972    );
973
974    // Execute query and map results
975    let rows = sqlx::query(&query)
976        .bind(instance_id)
977        .bind(status_filter)
978        .bind(&filter.step_type)
979        .bind(&filter.scope_id)
980        .bind(&filter.parent_scope_id)
981        .bind(filter.root_scopes_only)
982        .bind(limit)
983        .bind(offset)
984        .fetch_all(pool)
985        .await?;
986
987    // Map rows to StepSummaryRecord
988    let mut records = Vec::with_capacity(rows.len());
989    for row in rows {
990        use sqlx::Row;
991
992        let status_str: &str = row.get("status");
993        let status = match status_str {
994            "running" => StepStatus::Running,
995            "failed" => StepStatus::Failed,
996            _ => StepStatus::Completed,
997        };
998
999        records.push(StepSummaryRecord {
1000            step_id: row.get("step_id"),
1001            step_name: row.get("step_name"),
1002            step_type: row
1003                .get::<Option<String>, _>("step_type")
1004                .unwrap_or_default(),
1005            status,
1006            started_at: row.get("started_at"),
1007            completed_at: row.get("completed_at"),
1008            duration_ms: row.get("duration_ms"),
1009            inputs: row.get("inputs"),
1010            outputs: row.get("outputs"),
1011            error: row.get("error"),
1012            scope_id: row.get("scope_id"),
1013            parent_scope_id: row.get("parent_scope_id"),
1014        });
1015    }
1016
1017    Ok(records)
1018}
1019
1020/// Count step summaries for an instance with filtering.
1021pub async fn count_step_summaries(
1022    pool: &PgPool,
1023    instance_id: &str,
1024    filter: &ListStepSummariesFilter,
1025) -> Result<i64, CoreError> {
1026    // Convert status filter to string for SQL CASE matching
1027    let status_filter: Option<&str> = filter.status.map(|s| match s {
1028        StepStatus::Running => "running",
1029        StepStatus::Completed => "completed",
1030        StepStatus::Failed => "failed",
1031    });
1032
1033    let count: (i64,) = sqlx::query_as(
1034        r#"
1035        WITH start_events AS (
1036            SELECT
1037                convert_from(payload, 'UTF8')::jsonb->>'step_id' as step_id,
1038                convert_from(payload, 'UTF8')::jsonb->>'step_type' as step_type,
1039                convert_from(payload, 'UTF8')::jsonb->>'scope_id' as scope_id,
1040                convert_from(payload, 'UTF8')::jsonb->>'parent_scope_id' as parent_scope_id
1041            FROM instance_events
1042            WHERE instance_id = $1 AND subtype = 'step_debug_start'
1043        ),
1044        end_events AS (
1045            SELECT
1046                convert_from(payload, 'UTF8')::jsonb->>'step_id' as step_id,
1047                convert_from(payload, 'UTF8')::jsonb->>'scope_id' as scope_id,
1048                convert_from(payload, 'UTF8')::jsonb->'error' as error
1049            FROM instance_events
1050            WHERE instance_id = $1 AND subtype = 'step_debug_end'
1051        ),
1052        paired AS (
1053            SELECT
1054                s.step_id,
1055                s.step_type,
1056                s.scope_id,
1057                s.parent_scope_id,
1058                CASE
1059                    WHEN e.step_id IS NULL THEN 'running'
1060                    WHEN e.error IS NOT NULL AND e.error::text != 'null' THEN 'failed'
1061                    ELSE 'completed'
1062                END as status
1063            FROM start_events s
1064            LEFT JOIN end_events e ON s.step_id = e.step_id AND COALESCE(s.scope_id, '') = COALESCE(e.scope_id, '')
1065        )
1066        SELECT COUNT(*)
1067        FROM paired
1068        WHERE ($2::TEXT IS NULL OR status = $2)
1069          AND ($3::TEXT IS NULL OR step_type = $3)
1070          AND ($4::TEXT IS NULL OR scope_id = $4)
1071          AND ($5::TEXT IS NULL OR parent_scope_id = $5)
1072          AND (NOT $6 OR parent_scope_id IS NULL)
1073        "#,
1074    )
1075    .bind(instance_id)
1076    .bind(status_filter)
1077    .bind(&filter.step_type)
1078    .bind(&filter.scope_id)
1079    .bind(&filter.parent_scope_id)
1080    .bind(filter.root_scopes_only)
1081    .fetch_one(pool)
1082    .await?;
1083
1084    Ok(count.0)
1085}
1086
1087// ============================================================================
1088// Signal Operations
1089// ============================================================================
1090
1091/// Insert or update a pending signal.
1092/// Uses ON CONFLICT to replace existing signal for the same instance.
1093pub async fn insert_signal(
1094    pool: &PgPool,
1095    instance_id: &str,
1096    signal_type: &str,
1097    payload: &[u8],
1098) -> Result<(), CoreError> {
1099    let payload_opt = if payload.is_empty() {
1100        None
1101    } else {
1102        Some(payload)
1103    };
1104
1105    sqlx::query(
1106        r#"
1107        INSERT INTO pending_signals (instance_id, signal_type, payload, created_at)
1108        VALUES ($1, $2::signal_type, $3, NOW())
1109        ON CONFLICT (instance_id) DO UPDATE
1110        SET signal_type = EXCLUDED.signal_type,
1111            payload = EXCLUDED.payload,
1112            created_at = NOW(),
1113            acknowledged_at = NULL
1114        "#,
1115    )
1116    .bind(instance_id)
1117    .bind(signal_type)
1118    .bind(payload_opt)
1119    .execute(pool)
1120    .await?;
1121
1122    Ok(())
1123}
1124
1125/// Insert or update a pending custom signal scoped to a checkpoint.
1126pub async fn insert_custom_signal(
1127    pool: &PgPool,
1128    instance_id: &str,
1129    checkpoint_id: &str,
1130    payload: &[u8],
1131) -> Result<(), CoreError> {
1132    let payload_opt = if payload.is_empty() {
1133        None
1134    } else {
1135        Some(payload)
1136    };
1137
1138    sqlx::query(
1139        r#"
1140        INSERT INTO pending_checkpoint_signals (instance_id, checkpoint_id, payload, created_at)
1141        VALUES ($1, $2, $3, NOW())
1142        ON CONFLICT (instance_id, checkpoint_id) DO UPDATE
1143        SET payload = EXCLUDED.payload,
1144            created_at = NOW()
1145        "#,
1146    )
1147    .bind(instance_id)
1148    .bind(checkpoint_id)
1149    .bind(payload_opt)
1150    .execute(pool)
1151    .await?;
1152
1153    Ok(())
1154}
1155
1156/// Get pending signal for an instance (not yet acknowledged).
1157pub async fn get_pending_signal(
1158    pool: &PgPool,
1159    instance_id: &str,
1160) -> Result<Option<SignalRecord>, CoreError> {
1161    let record = sqlx::query_as::<_, SignalRecord>(
1162        r#"
1163        SELECT instance_id, signal_type::text as signal_type, payload, created_at, acknowledged_at
1164        FROM pending_signals
1165        WHERE instance_id = $1 AND acknowledged_at IS NULL
1166        "#,
1167    )
1168    .bind(instance_id)
1169    .fetch_optional(pool)
1170    .await?;
1171
1172    Ok(record)
1173}
1174
1175/// Take a pending custom signal for a checkpoint (delete and return it).
1176pub async fn take_pending_custom_signal(
1177    pool: &PgPool,
1178    instance_id: &str,
1179    checkpoint_id: &str,
1180) -> Result<Option<CustomSignalRecord>, CoreError> {
1181    let record = sqlx::query_as::<_, CustomSignalRecord>(
1182        r#"
1183        DELETE FROM pending_checkpoint_signals
1184        WHERE instance_id = $1 AND checkpoint_id = $2
1185        RETURNING instance_id, checkpoint_id, payload, created_at
1186        "#,
1187    )
1188    .bind(instance_id)
1189    .bind(checkpoint_id)
1190    .fetch_optional(pool)
1191    .await?;
1192
1193    Ok(record)
1194}
1195
1196/// Acknowledge a signal.
1197pub async fn acknowledge_signal(pool: &PgPool, instance_id: &str) -> Result<(), CoreError> {
1198    sqlx::query(
1199        r#"
1200        UPDATE pending_signals
1201        SET acknowledged_at = NOW()
1202        WHERE instance_id = $1 AND acknowledged_at IS NULL
1203        "#,
1204    )
1205    .bind(instance_id)
1206    .execute(pool)
1207    .await?;
1208
1209    Ok(())
1210}
1211
1212// ============================================================================
1213// Health Operations
1214// ============================================================================
1215
1216/// Count active instances (running or suspended).
1217pub async fn count_active_instances(pool: &PgPool) -> Result<i64, CoreError> {
1218    let row: (i64,) = sqlx::query_as(
1219        r#"
1220        SELECT COUNT(*)
1221        FROM instances
1222        WHERE status IN ('running', 'suspended')
1223        "#,
1224    )
1225    .fetch_one(pool)
1226    .await?;
1227
1228    Ok(row.0)
1229}
1230
1231// ============================================================================
1232// Sleep Operations
1233// ============================================================================
1234
1235/// Set the sleep_until timestamp for an instance.
1236pub async fn set_instance_sleep(
1237    pool: &PgPool,
1238    instance_id: &str,
1239    sleep_until: DateTime<Utc>,
1240) -> Result<(), CoreError> {
1241    let result = sqlx::query(
1242        r#"
1243        UPDATE instances
1244        SET sleep_until = $2
1245        WHERE instance_id = $1
1246        "#,
1247    )
1248    .bind(instance_id)
1249    .bind(sleep_until)
1250    .execute(pool)
1251    .await?;
1252
1253    if result.rows_affected() == 0 {
1254        return Err(CoreError::InstanceNotFound {
1255            instance_id: instance_id.to_string(),
1256        });
1257    }
1258
1259    Ok(())
1260}
1261
1262/// Clear the sleep_until timestamp for an instance.
1263pub async fn clear_instance_sleep(pool: &PgPool, instance_id: &str) -> Result<(), CoreError> {
1264    let result = sqlx::query(
1265        r#"
1266        UPDATE instances
1267        SET sleep_until = NULL
1268        WHERE instance_id = $1
1269        "#,
1270    )
1271    .bind(instance_id)
1272    .execute(pool)
1273    .await?;
1274
1275    if result.rows_affected() == 0 {
1276        return Err(CoreError::InstanceNotFound {
1277            instance_id: instance_id.to_string(),
1278        });
1279    }
1280
1281    Ok(())
1282}
1283
1284/// Get instances that are due to wake (sleep_until <= now).
1285pub async fn get_sleeping_instances_due(
1286    pool: &PgPool,
1287    limit: i64,
1288) -> Result<Vec<InstanceRecord>, CoreError> {
1289    let records = sqlx::query_as::<_, InstanceRecord>(
1290        r#"
1291        SELECT instance_id, tenant_id, definition_version,
1292               status::text as status, checkpoint_id, attempt, max_attempts,
1293               created_at, started_at, finished_at, output, error, sleep_until
1294        FROM instances
1295        WHERE sleep_until IS NOT NULL
1296          AND sleep_until <= NOW()
1297          AND status = 'suspended'
1298        ORDER BY sleep_until ASC
1299        LIMIT $1
1300        "#,
1301    )
1302    .bind(limit)
1303    .fetch_all(pool)
1304    .await?;
1305
1306    Ok(records)
1307}
1308
1309#[async_trait::async_trait]
1310impl Persistence for PostgresPersistence {
1311    async fn register_instance(&self, instance_id: &str, tenant_id: &str) -> Result<(), CoreError> {
1312        create_instance(&self.pool, instance_id, tenant_id).await
1313    }
1314
1315    async fn get_instance(&self, instance_id: &str) -> Result<Option<InstanceRecord>, CoreError> {
1316        get_instance(&self.pool, instance_id).await
1317    }
1318
1319    async fn update_instance_status(
1320        &self,
1321        instance_id: &str,
1322        status: &str,
1323        started_at: Option<DateTime<Utc>>,
1324    ) -> Result<(), CoreError> {
1325        update_instance_status(&self.pool, instance_id, status, started_at).await
1326    }
1327
1328    async fn update_instance_checkpoint(
1329        &self,
1330        instance_id: &str,
1331        checkpoint_id: &str,
1332    ) -> Result<(), CoreError> {
1333        update_instance_checkpoint(&self.pool, instance_id, checkpoint_id).await
1334    }
1335
1336    async fn complete_instance(
1337        &self,
1338        instance_id: &str,
1339        output: Option<&[u8]>,
1340        error: Option<&str>,
1341    ) -> Result<(), CoreError> {
1342        complete_instance(&self.pool, instance_id, output, error).await
1343    }
1344
1345    async fn save_checkpoint(
1346        &self,
1347        instance_id: &str,
1348        checkpoint_id: &str,
1349        state: &[u8],
1350    ) -> Result<(), CoreError> {
1351        save_checkpoint(&self.pool, instance_id, checkpoint_id, state).await
1352    }
1353
1354    async fn load_checkpoint(
1355        &self,
1356        instance_id: &str,
1357        checkpoint_id: &str,
1358    ) -> Result<Option<CheckpointRecord>, CoreError> {
1359        load_checkpoint(&self.pool, instance_id, checkpoint_id).await
1360    }
1361
1362    async fn list_checkpoints(
1363        &self,
1364        instance_id: &str,
1365        checkpoint_id: Option<&str>,
1366        limit: i64,
1367        offset: i64,
1368        created_after: Option<DateTime<Utc>>,
1369        created_before: Option<DateTime<Utc>>,
1370    ) -> Result<Vec<CheckpointRecord>, CoreError> {
1371        list_checkpoints(
1372            &self.pool,
1373            instance_id,
1374            checkpoint_id,
1375            limit,
1376            offset,
1377            created_after,
1378            created_before,
1379        )
1380        .await
1381    }
1382
1383    async fn count_checkpoints(
1384        &self,
1385        instance_id: &str,
1386        checkpoint_id: Option<&str>,
1387        created_after: Option<DateTime<Utc>>,
1388        created_before: Option<DateTime<Utc>>,
1389    ) -> Result<i64, CoreError> {
1390        count_checkpoints(
1391            &self.pool,
1392            instance_id,
1393            checkpoint_id,
1394            created_after,
1395            created_before,
1396        )
1397        .await
1398    }
1399
1400    async fn insert_event(&self, event: &EventRecord) -> Result<(), CoreError> {
1401        insert_event(&self.pool, event).await
1402    }
1403
1404    async fn insert_signal(
1405        &self,
1406        instance_id: &str,
1407        signal_type: &str,
1408        payload: &[u8],
1409    ) -> Result<(), CoreError> {
1410        insert_signal(&self.pool, instance_id, signal_type, payload).await
1411    }
1412
1413    async fn get_pending_signal(
1414        &self,
1415        instance_id: &str,
1416    ) -> Result<Option<SignalRecord>, CoreError> {
1417        get_pending_signal(&self.pool, instance_id).await
1418    }
1419
1420    async fn acknowledge_signal(&self, instance_id: &str) -> Result<(), CoreError> {
1421        acknowledge_signal(&self.pool, instance_id).await
1422    }
1423
1424    async fn insert_custom_signal(
1425        &self,
1426        instance_id: &str,
1427        checkpoint_id: &str,
1428        payload: &[u8],
1429    ) -> Result<(), CoreError> {
1430        insert_custom_signal(&self.pool, instance_id, checkpoint_id, payload).await
1431    }
1432
1433    async fn take_pending_custom_signal(
1434        &self,
1435        instance_id: &str,
1436        checkpoint_id: &str,
1437    ) -> Result<Option<CustomSignalRecord>, CoreError> {
1438        take_pending_custom_signal(&self.pool, instance_id, checkpoint_id).await
1439    }
1440
1441    async fn save_retry_attempt(
1442        &self,
1443        instance_id: &str,
1444        checkpoint_id: &str,
1445        attempt: i32,
1446        error_message: Option<&str>,
1447    ) -> Result<(), CoreError> {
1448        save_retry_attempt(
1449            &self.pool,
1450            instance_id,
1451            checkpoint_id,
1452            attempt,
1453            error_message,
1454        )
1455        .await
1456    }
1457
1458    async fn list_instances(
1459        &self,
1460        tenant_id: Option<&str>,
1461        status: Option<&str>,
1462        limit: i64,
1463        offset: i64,
1464    ) -> Result<Vec<InstanceRecord>, CoreError> {
1465        list_instances(&self.pool, tenant_id, status, limit, offset).await
1466    }
1467
1468    async fn health_check_db(&self) -> Result<bool, CoreError> {
1469        health_check_db(&self.pool).await
1470    }
1471
1472    async fn count_active_instances(&self) -> Result<i64, CoreError> {
1473        count_active_instances(&self.pool).await
1474    }
1475
1476    async fn set_instance_sleep(
1477        &self,
1478        instance_id: &str,
1479        sleep_until: DateTime<Utc>,
1480    ) -> Result<(), CoreError> {
1481        set_instance_sleep(&self.pool, instance_id, sleep_until).await
1482    }
1483
1484    async fn clear_instance_sleep(&self, instance_id: &str) -> Result<(), CoreError> {
1485        clear_instance_sleep(&self.pool, instance_id).await
1486    }
1487
1488    async fn get_sleeping_instances_due(
1489        &self,
1490        limit: i64,
1491    ) -> Result<Vec<InstanceRecord>, CoreError> {
1492        get_sleeping_instances_due(&self.pool, limit).await
1493    }
1494
1495    async fn list_events(
1496        &self,
1497        instance_id: &str,
1498        filter: &ListEventsFilter,
1499        limit: i64,
1500        offset: i64,
1501    ) -> Result<Vec<EventRecord>, CoreError> {
1502        list_events(&self.pool, instance_id, filter, limit, offset).await
1503    }
1504
1505    async fn count_events(
1506        &self,
1507        instance_id: &str,
1508        filter: &ListEventsFilter,
1509    ) -> Result<i64, CoreError> {
1510        count_events(&self.pool, instance_id, filter).await
1511    }
1512
1513    async fn list_step_summaries(
1514        &self,
1515        instance_id: &str,
1516        filter: &ListStepSummariesFilter,
1517        limit: i64,
1518        offset: i64,
1519    ) -> Result<Vec<StepSummaryRecord>, CoreError> {
1520        list_step_summaries(&self.pool, instance_id, filter, limit, offset).await
1521    }
1522
1523    async fn count_step_summaries(
1524        &self,
1525        instance_id: &str,
1526        filter: &ListStepSummariesFilter,
1527    ) -> Result<i64, CoreError> {
1528        count_step_summaries(&self.pool, instance_id, filter).await
1529    }
1530
1531    async fn complete_instance_extended(
1532        &self,
1533        instance_id: &str,
1534        status: &str,
1535        output: Option<&[u8]>,
1536        error: Option<&str>,
1537        stderr: Option<&str>,
1538        checkpoint_id: Option<&str>,
1539    ) -> Result<(), CoreError> {
1540        complete_instance_extended(
1541            &self.pool,
1542            instance_id,
1543            status,
1544            output,
1545            error,
1546            stderr,
1547            checkpoint_id,
1548        )
1549        .await
1550    }
1551
1552    async fn complete_instance_if_running(
1553        &self,
1554        instance_id: &str,
1555        status: &str,
1556        output: Option<&[u8]>,
1557        error: Option<&str>,
1558        stderr: Option<&str>,
1559        checkpoint_id: Option<&str>,
1560    ) -> Result<bool, CoreError> {
1561        complete_instance_if_running(
1562            &self.pool,
1563            instance_id,
1564            status,
1565            output,
1566            error,
1567            stderr,
1568            checkpoint_id,
1569        )
1570        .await
1571    }
1572
1573    async fn complete_instance_with_termination(
1574        &self,
1575        instance_id: &str,
1576        status: &str,
1577        termination_reason: Option<&str>,
1578        exit_code: Option<i32>,
1579        output: Option<&[u8]>,
1580        error: Option<&str>,
1581        stderr: Option<&str>,
1582        checkpoint_id: Option<&str>,
1583    ) -> Result<(), CoreError> {
1584        complete_instance_with_termination(
1585            &self.pool,
1586            instance_id,
1587            status,
1588            termination_reason,
1589            exit_code,
1590            output,
1591            error,
1592            stderr,
1593            checkpoint_id,
1594        )
1595        .await
1596    }
1597
1598    async fn complete_instance_with_termination_if_running(
1599        &self,
1600        instance_id: &str,
1601        status: &str,
1602        termination_reason: Option<&str>,
1603        exit_code: Option<i32>,
1604        output: Option<&[u8]>,
1605        error: Option<&str>,
1606        stderr: Option<&str>,
1607        checkpoint_id: Option<&str>,
1608    ) -> Result<bool, CoreError> {
1609        complete_instance_with_termination_if_running(
1610            &self.pool,
1611            instance_id,
1612            status,
1613            termination_reason,
1614            exit_code,
1615            output,
1616            error,
1617            stderr,
1618            checkpoint_id,
1619        )
1620        .await
1621    }
1622
1623    async fn update_instance_metrics(
1624        &self,
1625        instance_id: &str,
1626        memory_peak_bytes: Option<u64>,
1627        cpu_usage_usec: Option<u64>,
1628    ) -> Result<(), CoreError> {
1629        update_instance_metrics(&self.pool, instance_id, memory_peak_bytes, cpu_usage_usec).await
1630    }
1631
1632    async fn update_instance_stderr(
1633        &self,
1634        instance_id: &str,
1635        stderr: &str,
1636    ) -> Result<(), CoreError> {
1637        update_instance_stderr(&self.pool, instance_id, stderr).await
1638    }
1639
1640    async fn store_instance_input(&self, instance_id: &str, input: &[u8]) -> Result<(), CoreError> {
1641        store_instance_input(&self.pool, instance_id, input).await
1642    }
1643
1644    async fn get_terminal_instances_older_than(
1645        &self,
1646        older_than: DateTime<Utc>,
1647        limit: i64,
1648    ) -> Result<Vec<String>, CoreError> {
1649        get_terminal_instances_older_than(&self.pool, older_than, limit).await
1650    }
1651
1652    async fn delete_instances_batch(&self, instance_ids: &[String]) -> Result<u64, CoreError> {
1653        delete_instances_batch(&self.pool, instance_ids).await
1654    }
1655}
1656
1657/// Check database health.
1658pub async fn health_check_db(pool: &PgPool) -> Result<bool, CoreError> {
1659    let result: Result<(i32,), _> = sqlx::query_as("SELECT 1").fetch_one(pool).await;
1660    Ok(result.is_ok())
1661}
1662
1663/// Get terminal instance IDs older than the specified timestamp.
1664///
1665/// Only returns instances with terminal status: completed, failed, cancelled.
1666/// Returns instance IDs ordered by finished_at (oldest first) for batch processing.
1667pub async fn get_terminal_instances_older_than(
1668    pool: &PgPool,
1669    older_than: DateTime<Utc>,
1670    limit: i64,
1671) -> Result<Vec<String>, CoreError> {
1672    let rows: Vec<(String,)> = sqlx::query_as(
1673        r#"
1674        SELECT instance_id
1675        FROM instances
1676        WHERE status IN ('completed', 'failed', 'cancelled')
1677          AND finished_at IS NOT NULL
1678          AND finished_at < $1
1679        ORDER BY finished_at ASC
1680        LIMIT $2
1681        "#,
1682    )
1683    .bind(older_than)
1684    .bind(limit)
1685    .fetch_all(pool)
1686    .await?;
1687
1688    Ok(rows.into_iter().map(|(id,)| id).collect())
1689}
1690
1691/// Delete instances by their IDs.
1692///
1693/// This deletes from the instances table; child tables with ON DELETE CASCADE
1694/// are automatically cleaned up by the database.
1695///
1696/// Returns the count of deleted instances.
1697pub async fn delete_instances_batch(
1698    pool: &PgPool,
1699    instance_ids: &[String],
1700) -> Result<u64, CoreError> {
1701    if instance_ids.is_empty() {
1702        return Ok(0);
1703    }
1704
1705    let result = sqlx::query("DELETE FROM instances WHERE instance_id = ANY($1)")
1706        .bind(instance_ids)
1707        .execute(pool)
1708        .await?;
1709
1710    Ok(result.rows_affected())
1711}
1712
1713/// List instances with optional filtering.
1714pub async fn list_instances(
1715    pool: &PgPool,
1716    tenant_id: Option<&str>,
1717    status_filter: Option<&str>,
1718    limit: i64,
1719    offset: i64,
1720) -> Result<Vec<InstanceRecord>, CoreError> {
1721    let mut query = String::from(
1722        r#"
1723        SELECT instance_id, tenant_id, definition_version,
1724               status::text as status, checkpoint_id, attempt, max_attempts,
1725               created_at, started_at, finished_at, output, error, sleep_until
1726        FROM instances
1727        WHERE 1=1
1728        "#,
1729    );
1730
1731    let mut params: Vec<String> = Vec::new();
1732    let mut param_idx = 1;
1733
1734    if let Some(tid) = tenant_id {
1735        query.push_str(&format!(" AND tenant_id = ${}", param_idx));
1736        params.push(tid.to_string());
1737        param_idx += 1;
1738    }
1739
1740    if let Some(status) = status_filter {
1741        query.push_str(&format!(" AND status::text = ${}", param_idx));
1742        params.push(status.to_string());
1743        param_idx += 1;
1744    }
1745
1746    query.push_str(&format!(
1747        " ORDER BY created_at DESC LIMIT ${} OFFSET ${}",
1748        param_idx,
1749        param_idx + 1
1750    ));
1751
1752    // Build and execute the query dynamically
1753    let mut sqlx_query = sqlx::query_as::<_, InstanceRecord>(&query);
1754
1755    for param in &params {
1756        sqlx_query = sqlx_query.bind(param);
1757    }
1758    sqlx_query = sqlx_query.bind(limit).bind(offset);
1759
1760    let records = sqlx_query.fetch_all(pool).await?;
1761
1762    Ok(records)
1763}
1764
1765#[cfg(test)]
1766mod tests {
1767    use super::*;
1768
1769    use crate::migrations::POSTGRES as MIGRATOR;
1770
1771    // Helper to get a test database pool
1772    async fn test_pool() -> Option<PgPool> {
1773        let url = std::env::var("TEST_RUNTARA_DATABASE_URL").ok()?;
1774        let pool = PgPool::connect(&url).await.ok()?;
1775        MIGRATOR.run(&pool).await.ok()?;
1776        Some(pool)
1777    }
1778
1779    // Helper to create a test instance
1780    async fn create_test_instance(pool: &PgPool, instance_id: Uuid, tenant_id: &str) {
1781        sqlx::query(
1782            r#"
1783            INSERT INTO instances (instance_id, tenant_id, definition_version, status)
1784            VALUES ($1, $2, 1, 'pending')
1785            "#,
1786        )
1787        .bind(instance_id)
1788        .bind(tenant_id)
1789        .execute(pool)
1790        .await
1791        .expect("Failed to create test instance");
1792    }
1793
1794    // Helper to clean up test data
1795    async fn cleanup_test_instance(pool: &PgPool, instance_id: Uuid) {
1796        sqlx::query("DELETE FROM instances WHERE instance_id = $1")
1797            .bind(instance_id)
1798            .execute(pool)
1799            .await
1800            .ok();
1801    }
1802
1803    #[tokio::test]
1804    async fn test_insert_and_get_instance() {
1805        let Some(pool) = test_pool().await else {
1806            eprintln!("Skipping test: TEST_DATABASE_URL not set");
1807            return;
1808        };
1809
1810        let instance_id = Uuid::new_v4();
1811        create_test_instance(&pool, instance_id, "test-tenant").await;
1812
1813        let result = get_instance(&pool, &instance_id.to_string()).await;
1814        assert!(result.is_ok());
1815        let instance = result.unwrap();
1816        assert!(instance.is_some());
1817        let instance = instance.unwrap();
1818        assert_eq!(instance.tenant_id, "test-tenant");
1819        assert_eq!(instance.status, "pending");
1820
1821        cleanup_test_instance(&pool, instance_id).await;
1822    }
1823
1824    #[tokio::test]
1825    async fn test_update_instance_status() {
1826        let Some(pool) = test_pool().await else {
1827            eprintln!("Skipping test: TEST_DATABASE_URL not set");
1828            return;
1829        };
1830
1831        let instance_id = Uuid::new_v4();
1832        create_test_instance(&pool, instance_id, "test-tenant").await;
1833
1834        let result =
1835            update_instance_status(&pool, &instance_id.to_string(), "running", Some(Utc::now()))
1836                .await;
1837        assert!(result.is_ok());
1838
1839        let instance = get_instance(&pool, &instance_id.to_string())
1840            .await
1841            .unwrap()
1842            .unwrap();
1843        assert_eq!(instance.status, "running");
1844        assert!(instance.started_at.is_some());
1845
1846        cleanup_test_instance(&pool, instance_id).await;
1847    }
1848
1849    #[tokio::test]
1850    async fn test_update_instance_checkpoint() {
1851        let Some(pool) = test_pool().await else {
1852            eprintln!("Skipping test: TEST_DATABASE_URL not set");
1853            return;
1854        };
1855
1856        let instance_id = Uuid::new_v4();
1857        create_test_instance(&pool, instance_id, "test-tenant").await;
1858
1859        let result =
1860            update_instance_checkpoint(&pool, &instance_id.to_string(), "checkpoint-1").await;
1861        assert!(result.is_ok());
1862
1863        let instance = get_instance(&pool, &instance_id.to_string())
1864            .await
1865            .unwrap()
1866            .unwrap();
1867        assert_eq!(instance.checkpoint_id, Some("checkpoint-1".to_string()));
1868
1869        cleanup_test_instance(&pool, instance_id).await;
1870    }
1871
1872    #[tokio::test]
1873    async fn test_complete_instance_success() {
1874        let Some(pool) = test_pool().await else {
1875            eprintln!("Skipping test: TEST_DATABASE_URL not set");
1876            return;
1877        };
1878
1879        let instance_id = Uuid::new_v4();
1880        create_test_instance(&pool, instance_id, "test-tenant").await;
1881
1882        let output_data = b"success output";
1883        let result =
1884            complete_instance(&pool, &instance_id.to_string(), Some(output_data), None).await;
1885        assert!(result.is_ok());
1886
1887        let instance = get_instance(&pool, &instance_id.to_string())
1888            .await
1889            .unwrap()
1890            .unwrap();
1891        assert_eq!(instance.status, "completed");
1892        assert_eq!(instance.output, Some(output_data.to_vec()));
1893        assert!(instance.finished_at.is_some());
1894
1895        cleanup_test_instance(&pool, instance_id).await;
1896    }
1897
1898    #[tokio::test]
1899    async fn test_complete_instance_failure() {
1900        let Some(pool) = test_pool().await else {
1901            eprintln!("Skipping test: TEST_DATABASE_URL not set");
1902            return;
1903        };
1904
1905        let instance_id = Uuid::new_v4();
1906        create_test_instance(&pool, instance_id, "test-tenant").await;
1907
1908        let result =
1909            complete_instance(&pool, &instance_id.to_string(), None, Some("test error")).await;
1910        assert!(result.is_ok());
1911
1912        let instance = get_instance(&pool, &instance_id.to_string())
1913            .await
1914            .unwrap()
1915            .unwrap();
1916        assert_eq!(instance.status, "failed");
1917        assert_eq!(instance.error, Some("test error".to_string()));
1918        assert!(instance.finished_at.is_some());
1919
1920        cleanup_test_instance(&pool, instance_id).await;
1921    }
1922
1923    #[tokio::test]
1924    async fn test_save_checkpoint_new() {
1925        let Some(pool) = test_pool().await else {
1926            eprintln!("Skipping test: TEST_DATABASE_URL not set");
1927            return;
1928        };
1929
1930        let instance_id = Uuid::new_v4();
1931        create_test_instance(&pool, instance_id, "test-tenant").await;
1932
1933        let state = b"test state data";
1934        let result = save_checkpoint(&pool, &instance_id.to_string(), "cp-1", state).await;
1935        assert!(result.is_ok());
1936
1937        let checkpoint = load_checkpoint(&pool, &instance_id.to_string(), "cp-1")
1938            .await
1939            .unwrap();
1940        assert!(checkpoint.is_some());
1941        assert_eq!(checkpoint.unwrap().state, state.to_vec());
1942
1943        cleanup_test_instance(&pool, instance_id).await;
1944    }
1945
1946    #[tokio::test]
1947    async fn test_save_checkpoint_duplicate() {
1948        let Some(pool) = test_pool().await else {
1949            eprintln!("Skipping test: TEST_DATABASE_URL not set");
1950            return;
1951        };
1952
1953        let instance_id = Uuid::new_v4();
1954        create_test_instance(&pool, instance_id, "test-tenant").await;
1955
1956        // Save first checkpoint
1957        save_checkpoint(&pool, &instance_id.to_string(), "cp-1", b"state-1")
1958            .await
1959            .unwrap();
1960
1961        // Save again with same ID (should update)
1962        save_checkpoint(&pool, &instance_id.to_string(), "cp-1", b"state-2")
1963            .await
1964            .unwrap();
1965
1966        let checkpoint = load_checkpoint(&pool, &instance_id.to_string(), "cp-1")
1967            .await
1968            .unwrap()
1969            .unwrap();
1970        assert_eq!(checkpoint.state, b"state-2".to_vec());
1971
1972        cleanup_test_instance(&pool, instance_id).await;
1973    }
1974
1975    #[tokio::test]
1976    async fn test_load_checkpoint_by_id() {
1977        let Some(pool) = test_pool().await else {
1978            eprintln!("Skipping test: TEST_DATABASE_URL not set");
1979            return;
1980        };
1981
1982        let instance_id = Uuid::new_v4();
1983        create_test_instance(&pool, instance_id, "test-tenant").await;
1984
1985        save_checkpoint(&pool, &instance_id.to_string(), "cp-1", b"state-1")
1986            .await
1987            .unwrap();
1988        save_checkpoint(&pool, &instance_id.to_string(), "cp-2", b"state-2")
1989            .await
1990            .unwrap();
1991
1992        let cp1 = load_checkpoint(&pool, &instance_id.to_string(), "cp-1")
1993            .await
1994            .unwrap()
1995            .unwrap();
1996        assert_eq!(cp1.state, b"state-1".to_vec());
1997
1998        let cp2 = load_checkpoint(&pool, &instance_id.to_string(), "cp-2")
1999            .await
2000            .unwrap()
2001            .unwrap();
2002        assert_eq!(cp2.state, b"state-2".to_vec());
2003
2004        cleanup_test_instance(&pool, instance_id).await;
2005    }
2006
2007    #[tokio::test]
2008    async fn test_load_checkpoint_latest() {
2009        let Some(pool) = test_pool().await else {
2010            eprintln!("Skipping test: TEST_DATABASE_URL not set");
2011            return;
2012        };
2013
2014        let instance_id = Uuid::new_v4();
2015        create_test_instance(&pool, instance_id, "test-tenant").await;
2016
2017        save_checkpoint(&pool, &instance_id.to_string(), "cp-1", b"state-1")
2018            .await
2019            .unwrap();
2020        // Small delay to ensure different timestamps
2021        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
2022        save_checkpoint(&pool, &instance_id.to_string(), "cp-2", b"state-2")
2023            .await
2024            .unwrap();
2025        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
2026        save_checkpoint(&pool, &instance_id.to_string(), "cp-3", b"state-3")
2027            .await
2028            .unwrap();
2029
2030        let latest = load_latest_checkpoint(&pool, &instance_id.to_string())
2031            .await
2032            .unwrap()
2033            .unwrap();
2034        assert_eq!(latest.checkpoint_id, "cp-3");
2035        assert_eq!(latest.state, b"state-3".to_vec());
2036
2037        cleanup_test_instance(&pool, instance_id).await;
2038    }
2039
2040    #[tokio::test]
2041    async fn test_load_checkpoint_not_found() {
2042        let Some(pool) = test_pool().await else {
2043            eprintln!("Skipping test: TEST_DATABASE_URL not set");
2044            return;
2045        };
2046
2047        let instance_id = Uuid::new_v4();
2048        create_test_instance(&pool, instance_id, "test-tenant").await;
2049
2050        let result = load_checkpoint(&pool, &instance_id.to_string(), "nonexistent")
2051            .await
2052            .unwrap();
2053        assert!(result.is_none());
2054
2055        cleanup_test_instance(&pool, instance_id).await;
2056    }
2057
2058    // NOTE: Legacy wake_queue tests removed - wake scheduling now uses sleep_until column
2059    // on the instances table (see migration 003_drop_wake_queue.sql)
2060
2061    #[tokio::test]
2062    async fn test_insert_event() {
2063        let Some(pool) = test_pool().await else {
2064            eprintln!("Skipping test: TEST_DATABASE_URL not set");
2065            return;
2066        };
2067
2068        let instance_id = Uuid::new_v4();
2069        create_test_instance(&pool, instance_id, "test-tenant").await;
2070
2071        let event = EventRecord {
2072            id: None,
2073            instance_id: instance_id.to_string(),
2074            event_type: "started".to_string(),
2075            checkpoint_id: None,
2076            payload: None,
2077            created_at: Utc::now(),
2078            subtype: None,
2079        };
2080
2081        let result = insert_event(&pool, &event).await;
2082        assert!(result.is_ok());
2083
2084        // Verify event was inserted
2085        let count: (i64,) =
2086            sqlx::query_as("SELECT COUNT(*) FROM instance_events WHERE instance_id = $1")
2087                .bind(instance_id.to_string())
2088                .fetch_one(&pool)
2089                .await
2090                .unwrap();
2091        assert_eq!(count.0, 1);
2092
2093        cleanup_test_instance(&pool, instance_id).await;
2094    }
2095
2096    #[tokio::test]
2097    async fn test_insert_signal() {
2098        let Some(pool) = test_pool().await else {
2099            eprintln!("Skipping test: TEST_DATABASE_URL not set");
2100            return;
2101        };
2102
2103        let instance_id = Uuid::new_v4();
2104        create_test_instance(&pool, instance_id, "test-tenant").await;
2105
2106        let result = insert_signal(&pool, &instance_id.to_string(), "cancel", b"reason").await;
2107        assert!(result.is_ok());
2108
2109        let signal = get_pending_signal(&pool, &instance_id.to_string())
2110            .await
2111            .unwrap();
2112        assert!(signal.is_some());
2113        let signal = signal.unwrap();
2114        assert_eq!(signal.signal_type, "cancel");
2115        assert_eq!(signal.payload, Some(b"reason".to_vec()));
2116
2117        cleanup_test_instance(&pool, instance_id).await;
2118    }
2119
2120    #[tokio::test]
2121    async fn test_get_pending_signal() {
2122        let Some(pool) = test_pool().await else {
2123            eprintln!("Skipping test: TEST_DATABASE_URL not set");
2124            return;
2125        };
2126
2127        let instance_id = Uuid::new_v4();
2128        create_test_instance(&pool, instance_id, "test-tenant").await;
2129
2130        insert_signal(&pool, &instance_id.to_string(), "pause", b"")
2131            .await
2132            .unwrap();
2133
2134        let signal = get_pending_signal(&pool, &instance_id.to_string())
2135            .await
2136            .unwrap();
2137        assert!(signal.is_some());
2138        assert_eq!(signal.unwrap().signal_type, "pause");
2139
2140        cleanup_test_instance(&pool, instance_id).await;
2141    }
2142
2143    #[tokio::test]
2144    async fn test_get_pending_signal_none() {
2145        let Some(pool) = test_pool().await else {
2146            eprintln!("Skipping test: TEST_DATABASE_URL not set");
2147            return;
2148        };
2149
2150        let instance_id = Uuid::new_v4();
2151        create_test_instance(&pool, instance_id, "test-tenant").await;
2152
2153        let signal = get_pending_signal(&pool, &instance_id.to_string())
2154            .await
2155            .unwrap();
2156        assert!(signal.is_none());
2157
2158        cleanup_test_instance(&pool, instance_id).await;
2159    }
2160
2161    #[tokio::test]
2162    async fn test_acknowledge_signal() {
2163        let Some(pool) = test_pool().await else {
2164            eprintln!("Skipping test: TEST_DATABASE_URL not set");
2165            return;
2166        };
2167
2168        let instance_id = Uuid::new_v4();
2169        create_test_instance(&pool, instance_id, "test-tenant").await;
2170
2171        insert_signal(&pool, &instance_id.to_string(), "cancel", b"")
2172            .await
2173            .unwrap();
2174        acknowledge_signal(&pool, &instance_id.to_string())
2175            .await
2176            .unwrap();
2177
2178        // Should no longer return as pending
2179        let signal = get_pending_signal(&pool, &instance_id.to_string())
2180            .await
2181            .unwrap();
2182        assert!(signal.is_none());
2183
2184        cleanup_test_instance(&pool, instance_id).await;
2185    }
2186
2187    #[tokio::test]
2188    async fn test_insert_and_take_custom_signal() {
2189        let Some(pool) = test_pool().await else {
2190            eprintln!("Skipping test: TEST_DATABASE_URL not set");
2191            return;
2192        };
2193
2194        let instance_id = Uuid::new_v4();
2195        create_test_instance(&pool, instance_id, "test-tenant").await;
2196
2197        insert_custom_signal(&pool, &instance_id.to_string(), "wait-1", b"custom-payload")
2198            .await
2199            .unwrap();
2200
2201        // First take should retrieve and delete
2202        let signal = take_pending_custom_signal(&pool, &instance_id.to_string(), "wait-1")
2203            .await
2204            .unwrap()
2205            .expect("custom signal should exist");
2206        assert_eq!(signal.checkpoint_id, "wait-1");
2207        assert_eq!(signal.payload.unwrap(), b"custom-payload".to_vec());
2208
2209        // Second take should return none
2210        let signal = take_pending_custom_signal(&pool, &instance_id.to_string(), "wait-1")
2211            .await
2212            .unwrap();
2213        assert!(signal.is_none());
2214
2215        cleanup_test_instance(&pool, instance_id).await;
2216    }
2217
2218    #[tokio::test]
2219    async fn test_count_active_instances() {
2220        let Some(pool) = test_pool().await else {
2221            eprintln!("Skipping test: TEST_DATABASE_URL not set");
2222            return;
2223        };
2224
2225        let instance1 = Uuid::new_v4();
2226        let instance2 = Uuid::new_v4();
2227        create_test_instance(&pool, instance1, "test-tenant").await;
2228        create_test_instance(&pool, instance2, "test-tenant").await;
2229
2230        // Set one to running, one to suspended
2231        update_instance_status(&pool, &instance1.to_string(), "running", None)
2232            .await
2233            .unwrap();
2234        update_instance_status(&pool, &instance2.to_string(), "suspended", None)
2235            .await
2236            .unwrap();
2237
2238        let count = count_active_instances(&pool).await.unwrap();
2239        assert!(count >= 2); // At least our 2 test instances
2240
2241        cleanup_test_instance(&pool, instance1).await;
2242        cleanup_test_instance(&pool, instance2).await;
2243    }
2244
2245    #[tokio::test]
2246    async fn test_health_check_db() {
2247        let Some(pool) = test_pool().await else {
2248            eprintln!("Skipping test: TEST_DATABASE_URL not set");
2249            return;
2250        };
2251
2252        let result = health_check_db(&pool).await;
2253        assert!(result.is_ok());
2254        assert!(result.unwrap());
2255    }
2256}
runtara_core/persistence/postgres.rs

runtara_core/persistence/
postgres.rs
