haste_repository/pg/
failed_indexing.rs

1use crate::{
2    failed_indexing::{FailedIndexEntry, FailedIndexRecord, FailedIndexingProvider},
3    pg::{PGConnection, StoreError},
4};
5use haste_fhir_operation_error::OperationOutcomeError;
6use haste_jwt::{ProjectId, TenantId};
7use sqlx::{Acquire, PgExecutor, Postgres, QueryBuilder};
8
9async fn search_failed_indexing<'a, 'e, E>(
10    executor: E,
11    tenant: &'a TenantId,
12    project: &'a ProjectId,
13) -> Result<Vec<FailedIndexEntry>, OperationOutcomeError>
14where
15    E: PgExecutor<'e>,
16{
17    let entries = sqlx::query_as::<_, FailedIndexEntry>(
18        r"
19            SELECT r.id, f.version_id, f.resource_type, f.fhir_method, f.attempt_count,
20                   f.error_message, f.first_failed_at, f.last_failed_at, f.resolved_at,
21                   r.sequence
22            FROM failed_search_indexing f
23            JOIN resources r
24                ON r.tenant = f.tenant AND r.project = f.project AND r.version_id = f.version_id
25            WHERE f.tenant = $1 AND f.project = $2
26            ORDER BY f.last_failed_at DESC
27        ",
28    )
29    .bind(tenant.as_ref())
30    .bind(project.as_ref())
31    .fetch_all(executor)
32    .await
33    .map_err(StoreError::from)?;
34
35    Ok(entries)
36}
37
38impl FailedIndexingProvider for PGConnection {
39    async fn record_failures(
40        &self,
41        failures: &[FailedIndexRecord],
42    ) -> Result<(), OperationOutcomeError> {
43        if failures.is_empty() {
44            return Ok(());
45        }
46
47        match self {
48            PGConnection::Transaction(tx, _, _) => {
49                let mut tx = tx.lock().await;
50                let conn = (&mut (*tx)).acquire().await.map_err(StoreError::from)?;
51
52                let mut query_builder: QueryBuilder<Postgres> = QueryBuilder::new(
53                    "INSERT INTO failed_search_indexing \
54                     (tenant, project, version_id, resource_type, fhir_method, error_message) ",
55                );
56
57                query_builder.push_values(failures, |mut b, failure| {
58                    b.push_bind(failure.tenant.as_ref())
59                        .push_bind(failure.project.as_ref())
60                        .push_bind(failure.version_id.as_ref())
61                        .push_bind(&failure.resource_type)
62                        .push_bind(failure.fhir_method.clone())
63                        .push_bind(&failure.error_message);
64                });
65
66                query_builder.push(
67                    r" ON CONFLICT (tenant, project, version_id) DO UPDATE SET
68                      attempt_count = failed_search_indexing.attempt_count + 1,
69                      last_failed_at = now(),
70                      error_message = EXCLUDED.error_message,
71                      resolved_at = NULL",
72                );
73
74                query_builder
75                    .build()
76                    .execute(conn)
77                    .await
78                    .map_err(StoreError::from)?;
79
80                Ok(())
81            }
82            PGConnection::Pool(..) => Err(StoreError::NotTransaction.into()),
83        }
84    }
85
86    async fn search(
87        &self,
88        tenant: &TenantId,
89        project: &ProjectId,
90    ) -> Result<Vec<FailedIndexEntry>, OperationOutcomeError> {
91        match self {
92            PGConnection::Pool(pool, _) => search_failed_indexing(pool, tenant, project).await,
93            PGConnection::Transaction(tx, _, _) => {
94                let mut tx = tx.lock().await;
95                search_failed_indexing(&mut **tx, tenant, project).await
96            }
97        }
98    }
99}
haste_repository/pg/failed_indexing.rs

haste_repository/pg/
failed_indexing.rs
