Skip to content

Split ingestion profile default storage #266

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yvette-carlisle merged 1 commit into from
Jun 29, 2026
Merged

Split ingestion profile default storage #266

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
147 changes: 7 additions & 140 deletions packages/elf-service/src/ingestion_profiles/storage.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,13 +1,16 @@
mod defaults;

pub(super) use self::defaults::{
seed_default_profile, select_default_row, select_default_selector, upsert_default_row,
};

use serde_json::Value;
use sqlx::{FromRow, PgPool};
use time::OffsetDateTime;

use crate::{
Error, Result,
ingestion_profiles::{
ADD_EVENT_PIPELINE, DEFAULT_PROFILE_ID, DEFAULT_PROFILE_VERSION, profile,
types::IngestionProfileSelector,
},
ingestion_profiles::{ADD_EVENT_PIPELINE, types::IngestionProfileSelector},
};

#[derive(FromRow)]
Expand All @@ -34,13 +37,6 @@ pub(super) struct ProfileSummaryRow {
pub(super) created_by: String,
}

#[derive(FromRow)]
pub(super) struct ProfileDefaultRow {
pub(super) profile_id: String,
pub(super) version: Option<i32>,
pub(super) updated_at: OffsetDateTime,
}

pub(super) async fn next_profile_version(
pool: &PgPool,
tenant_id: &str,
Expand Down Expand Up @@ -186,59 +182,6 @@ ORDER BY version DESC",
Ok(rows)
}

pub(super) async fn select_default_row(
pool: &PgPool,
tenant_id: &str,
project_id: &str,
) -> Result<Option<ProfileDefaultRow>> {
let row = sqlx::query_as::<_, ProfileDefaultRow>(
"\
SELECT profile_id, version, updated_at
FROM memory_ingestion_profile_defaults
WHERE tenant_id=$1 AND project_id=$2 AND pipeline=$3",
)
.bind(tenant_id)
.bind(project_id)
.bind(ADD_EVENT_PIPELINE)
.fetch_optional(pool)
.await?;

Ok(row)
}

pub(super) async fn upsert_default_row(
pool: &PgPool,
tenant_id: &str,
project_id: &str,
profile_id: String,
version: i32,
) -> Result<ProfileDefaultRow> {
let row = sqlx::query_as::<_, ProfileDefaultRow>(
"\
INSERT INTO memory_ingestion_profile_defaults (
tenant_id,
project_id,
pipeline,
profile_id,
version
) VALUES ($1,$2,$3,$4,$5)
ON CONFLICT (tenant_id, project_id, pipeline) DO UPDATE
SET profile_id = EXCLUDED.profile_id,
version = EXCLUDED.version,
updated_at = now()
RETURNING profile_id, version, updated_at",
)
.bind(tenant_id)
.bind(project_id)
.bind(ADD_EVENT_PIPELINE)
.bind(profile_id)
.bind(version)
.fetch_one(pool)
.await?;

Ok(row)
}

pub(super) async fn select_profile(
pool: &PgPool,
tenant_id: &str,
Expand Down Expand Up @@ -283,79 +226,3 @@ LIMIT 1",
),
})
}

pub(super) async fn select_default_selector(
pool: &PgPool,
tenant_id: &str,
project_id: &str,
) -> Result<IngestionProfileSelector> {
let row = sqlx::query_as::<_, (String, Option<i32>)>(
"SELECT profile_id, version FROM memory_ingestion_profile_defaults WHERE tenant_id=$1 AND project_id=$2 AND pipeline=$3",
)
.bind(tenant_id)
.bind(project_id)
.bind(ADD_EVENT_PIPELINE)
.fetch_optional(pool)
.await?;
let row = match row {
Some((profile_id, version)) => IngestionProfileSelector { id: profile_id, version },
None => IngestionProfileSelector {
id: DEFAULT_PROFILE_ID.to_string(),
version: Some(DEFAULT_PROFILE_VERSION),
},
};

Ok(row)
}

pub(super) async fn seed_default_profile(
pool: &PgPool,
tenant_id: &str,
project_id: &str,
) -> Result<()> {
let profile =
serde_json::to_value(profile::builtin_profile_v1()).map_err(|_| Error::InvalidRequest {
message: "Failed to serialize default ingestion profile.".to_string(),
})?;

sqlx::query(
"\
INSERT INTO memory_ingestion_profiles (
tenant_id,
project_id,
pipeline,
profile_id,
version,
profile
) VALUES ($1,$2,$3,$4,$5,$6::jsonb)
ON CONFLICT DO NOTHING",
)
.bind(tenant_id)
.bind(project_id)
.bind(ADD_EVENT_PIPELINE)
.bind(DEFAULT_PROFILE_ID)
.bind(DEFAULT_PROFILE_VERSION)
.bind(profile)
.execute(pool)
.await?;
sqlx::query(
"\
INSERT INTO memory_ingestion_profile_defaults (
tenant_id,
project_id,
pipeline,
profile_id,
version
) VALUES ($1,$2,$3,$4,$5)
ON CONFLICT DO NOTHING",
)
.bind(tenant_id)
.bind(project_id)
.bind(ADD_EVENT_PIPELINE)
.bind(DEFAULT_PROFILE_ID)
.bind(DEFAULT_PROFILE_VERSION)
.execute(pool)
.await?;

Ok(())
}
146 changes: 146 additions & 0 deletions packages/elf-service/src/ingestion_profiles/storage/defaults.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,146 @@
use sqlx::{FromRow, PgPool};
use time::OffsetDateTime;

use crate::{
Error, Result,
ingestion_profiles::{
ADD_EVENT_PIPELINE, DEFAULT_PROFILE_ID, DEFAULT_PROFILE_VERSION, profile,
types::IngestionProfileSelector,
},
};

#[derive(FromRow)]
pub(in crate::ingestion_profiles) struct ProfileDefaultRow {
pub(in crate::ingestion_profiles) profile_id: String,
pub(in crate::ingestion_profiles) version: Option<i32>,
pub(in crate::ingestion_profiles) updated_at: OffsetDateTime,
}

pub(in crate::ingestion_profiles) async fn select_default_row(
pool: &PgPool,
tenant_id: &str,
project_id: &str,
) -> Result<Option<ProfileDefaultRow>> {
let row = sqlx::query_as::<_, ProfileDefaultRow>(
"\
SELECT profile_id, version, updated_at
FROM memory_ingestion_profile_defaults
WHERE tenant_id=$1 AND project_id=$2 AND pipeline=$3",
)
.bind(tenant_id)
.bind(project_id)
.bind(ADD_EVENT_PIPELINE)
.fetch_optional(pool)
.await?;

Ok(row)
}

pub(in crate::ingestion_profiles) async fn upsert_default_row(
pool: &PgPool,
tenant_id: &str,
project_id: &str,
profile_id: String,
version: i32,
) -> Result<ProfileDefaultRow> {
let row = sqlx::query_as::<_, ProfileDefaultRow>(
"\
INSERT INTO memory_ingestion_profile_defaults (
tenant_id,
project_id,
pipeline,
profile_id,
version
) VALUES ($1,$2,$3,$4,$5)
ON CONFLICT (tenant_id, project_id, pipeline) DO UPDATE
SET profile_id = EXCLUDED.profile_id,
version = EXCLUDED.version,
updated_at = now()
RETURNING profile_id, version, updated_at",
)
.bind(tenant_id)
.bind(project_id)
.bind(ADD_EVENT_PIPELINE)
.bind(profile_id)
.bind(version)
.fetch_one(pool)
.await?;

Ok(row)
}

pub(in crate::ingestion_profiles) async fn select_default_selector(
pool: &PgPool,
tenant_id: &str,
project_id: &str,
) -> Result<IngestionProfileSelector> {
let row = sqlx::query_as::<_, (String, Option<i32>)>(
"SELECT profile_id, version FROM memory_ingestion_profile_defaults WHERE tenant_id=$1 AND project_id=$2 AND pipeline=$3",
)
.bind(tenant_id)
.bind(project_id)
.bind(ADD_EVENT_PIPELINE)
.fetch_optional(pool)
.await?;
let row = match row {
Some((profile_id, version)) => IngestionProfileSelector { id: profile_id, version },
None => IngestionProfileSelector {
id: DEFAULT_PROFILE_ID.to_string(),
version: Some(DEFAULT_PROFILE_VERSION),
},
};

Ok(row)
}

pub(in crate::ingestion_profiles) async fn seed_default_profile(
pool: &PgPool,
tenant_id: &str,
project_id: &str,
) -> Result<()> {
let profile =
serde_json::to_value(profile::builtin_profile_v1()).map_err(|_| Error::InvalidRequest {
message: "Failed to serialize default ingestion profile.".to_string(),
})?;

sqlx::query(
"\
INSERT INTO memory_ingestion_profiles (
tenant_id,
project_id,
pipeline,
profile_id,
version,
profile
) VALUES ($1,$2,$3,$4,$5,$6::jsonb)
ON CONFLICT DO NOTHING",
)
.bind(tenant_id)
.bind(project_id)
.bind(ADD_EVENT_PIPELINE)
.bind(DEFAULT_PROFILE_ID)
.bind(DEFAULT_PROFILE_VERSION)
.bind(profile)
.execute(pool)
.await?;
sqlx::query(
"\
INSERT INTO memory_ingestion_profile_defaults (
tenant_id,
project_id,
pipeline,
profile_id,
version
) VALUES ($1,$2,$3,$4,$5)
ON CONFLICT DO NOTHING",
)
.bind(tenant_id)
.bind(project_id)
.bind(ADD_EVENT_PIPELINE)
.bind(DEFAULT_PROFILE_ID)
.bind(DEFAULT_PROFILE_VERSION)
.execute(pool)
.await?;

Ok(())
}