Resources

// build.rs
fn main() {
    valence_codegen::build().expect("valence codegen failed");
}

// schemas/counter_valence_schema.rs
use valence::prelude::*;

valence_schema! {
    Counter {
        repository: "https://github.com/acme/counter",
        table: "counter",
        version: "0.1.0",
        description: "Simple counter",
        // No policies block: every actor except System is denied.
        fields: [
            id: { r#type: FieldType::String, primary_key: true, required: true },
            value: { r#type: FieldType::Integer, required: true },
        ],
    }
}

// src/lib.rs
use valence::{use_, Model};
valence::include_generated_models!();

let counter = Counter::get(
    "home",
    &valence,
    use_!("Show the **counter value** on the home page."),
)
.await?;

use std::sync::Arc;
use boson::{
    configure, task, Boson, ExecutionContext, JsonExecutionContextFactory,
    MemQueueBackend,
};

#[task(name = "process_order", max_attempts = 5, max_in_flight = 20)]
async fn process_order(
    ctx: Box<dyn ExecutionContext>,
    order_id: String,
    amount_cents: u64,
) -> boson_core::Result<()> {
    tracing::info!(actor = ctx.label(), %order_id, amount_cents);
    Ok(())
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let boson = Boson::builder()
        .queue_backend(Arc::new(MemQueueBackend::new()))
        .execution_context_factory(JsonExecutionContextFactory)
        .auto_registry()
        .build()?;
    configure(boson);

    ProcessOrder::send_with(
        serde_json::json!({"System": {"operation": "checkout"}}),
        ProcessOrderParams {
            order_id: "ord-42".into(),
            amount_cents: 9900,
        },
    )
    .await?;
    Ok(())
}

use std::sync::Arc;
use chronon::prelude::*;
use chronon::InMemorySchedulerStore;

#[chronon::script(name = "nightly_cleanup")]
async fn nightly_cleanup(
    ctx: Box<dyn ScriptContext>,
    retention_days: u32,
) -> chronon::Result<()> {
    println!("{}: retaining {retention_days} days", ctx.label());
    Ok(())
}

#[tokio::main]
async fn main() -> chronon::Result<()> {
    let chronon = Chronon::builder()
        .scheduler_store(Arc::new(InMemorySchedulerStore::new()))
        .context_factory(Arc::new(JsonScriptContextFactory))
        .embedded()
        .auto_registry()
        .build()?;

    let nightly = JobBuilder::new(&nightly_cleanup())
        .name("nightly-schedule")
        .cron("0 2 * * *")?
        .timezone("UTC")
        .params(NightlyCleanupParams { retention_days: 7 })
        .build()?;
    chronon.coordinator_service().upsert_job(nightly).await?;

    chronon.run().await
}

use std::sync::Arc;
use photon::{subscribe, topic, JsonIdentityFactory, Photon};
use photon_core::Actor;

#[topic(name = "orders.created", keyed_by = "order_id")]
pub struct OrderCreated {
    pub order_id: String,
    pub amount_cents: u64,
}

#[subscribe(topic = "orders.created", durable = "billing")]
async fn on_order_created(actor: Box<dyn Actor>, event: OrderCreated) -> photon::Result<()> {
    tracing::info!(actor = actor.label(), order = %event.order_id, cents = event.amount_cents);
    Ok(())
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // Requires PHOTON_TRANSPORT_KEY (base64, 32 bytes).
    let photon = Photon::builder().auto_registry().build()?;
    photon.start_executor(Arc::new(JsonIdentityFactory))?;

    OrderCreated {
        order_id: "ord-1".into(),
        amount_cents: 9900,
    }
    .publish_on(&photon)
    .await?;
    Ok(())
}

use spectra_macros::{spectra_metric, spectra_schema};

spectra_schema! {
    RequestDebugLog {
        store: "default",
        table: "request_debug_log",
        version: "0.1.0",
        description: "Structured debug events for request tracing",
        fields: [
            message: {
                r#type: String,
                classification: { pii: false, safe_for_console: true },
            },
        ],
    }
}

spectra_metric! {
    CacheHits {
        store: "default",
        name: "cache_hits",
        version: "0.1.0",
        description: "Counter for cache hit events",
    }
}

use gluon::applications::{
    ApplicationService, ContainerConfigSpec, CreateApplicationParams, RouteConfigSpec,
};
use gluon::domain::attach_hostname;
use gluon::generated::{GluonApplicationAppKind, GluonDomainZone};
use valence::{use_, Model};

let booted = gluon::bootstrap_valence_from_env().await?;
let valence = booted.valence("deploy-marketing")?;

let app = ApplicationService::create(
    &valence,
    CreateApplicationParams {
        name: "marketing".into(),
        image_ref: "ghcr.io/acme/marketing:latest".into(),
        app_kind: GluonApplicationAppKind::Service,
        container: ContainerConfigSpec::default(),
        health_check: None,
        lb: None,
        route: Some(RouteConfigSpec { path_prefix: "/".into(), strip_prefix: false }),
        desired_instances: 2,
        target_node_id: None,
        metadata_json: serde_json::json!({}),
    },
)
.await?;

let zone = GluonDomainZone::get(
    "zone:example.com",
    &valence,
    use_!("Load the **DNS zone** the new hostname will live in."),
)
.await?
.expect("zone");
attach_hostname(&valence, &app, &zone, "app.example.com").await?;

use nucleus::endpoint::resolve_db_endpoint;
use nucleus::provision::{provision, EngineStrategy, ProvisionPolicy};
use nucleus::reconcile::reconcile_stack;
use nucleus::shards::add_shard;

// us_east: the GluonCellCloudPlacement row for that region
let stack = provision(
    valence,
    EngineStrategy::Postgres,
    ProvisionPolicy {
        logical_name: "primary".into(),
        replica_count: 1,
    },
    Some(&us_east),
)
.await?;

reconcile_stack(valence, &stack).await?;
let url = resolve_db_endpoint(valence, "primary").await?;

// Later, when it needs to grow:
add_shard(valence, &stack).await?;

use leptos::prelude::*;
use orbital::components::{
    Button, Card, CardContent, CardFooter, CardHeader, Subtitle1,
};
use orbital::OrbitalTemplate;

#[component]
fn App() -> impl IntoView {
    view! {
        <OrbitalTemplate>
            <Card>
                <CardHeader>
                    <Subtitle1>"Account"</Subtitle1>
                </CardHeader>
                <CardContent>"Signed-in settings use the same tokens as marketing."</CardContent>
                <CardFooter>
                    <Button>"Save"</Button>
                </CardFooter>
            </Card>
        </OrbitalTemplate>
    }
}

use boson_core::ExecutionContext;
use boson_macros::task;
use boson_valence_identity::valence_from_context;
use higgs::Higgs;
use leptos::prelude::*;

// `auth` rejects callers who aren't signed in.
#[higgs_macros::server(auth)]
pub async fn queue_profile_rebuild() -> Result<(), ServerFnError> {
    let ctx = Higgs::from_request().await?;
    let valence = ctx.valence().map_err(ServerFnError::new)?;
    let user_id = valence
        .actor()
        .user_id()
        .ok_or_else(|| ServerFnError::new("expected a signed-in user"))?
        .to_string();

    RebuildProfile::send_with(
        serde_json::json!({ "User": { "user_id": user_id } }),
        RebuildProfileParams {
            user_id: user_id.clone(),
        },
    )
    .await
    .map_err(ServerFnError::new)?;
    Ok(())
}

#[task(name = "rebuild_profile")]
async fn rebuild_profile(
    ctx: Box<dyn ExecutionContext>,
    user_id: String,
) -> boson_core::Result<()> {
    // Same user as the server function. Chronon scripts and Photon
    // subscribers rebuild context from the same ValenceFactory.
    let valence = valence_from_context(ctx.as_ref())?;
    tracing::info!(rebuilt_user = valence.actor().user_id(), %user_id);
    Ok(())
}

use neutrino::secret_store::{PutSecretRequest, SecretStore};
use neutrino::{create_initial_neutrino_groups, ValenceSealedStore};
use std::sync::Arc;
use valence::Actor;

create_initial_neutrino_groups(&valence).await?;
let store = ValenceSealedStore {
    valence: Arc::new(valence.with_actor(Actor::System {
        operation: "seal_smtp".into(),
    })),
    request_actor: Some("service:smtp-boot".into()),
};
let secret_ref = store
    .put(PutSecretRequest {
        name: "smtp_password".into(),
        scope_path: "/uf-notifications/smtp".into(),
        kind: "password".into(),
        plaintext: b"correct-horse-battery-staple".to_vec(),
        owner_actor: "service:smtp-boot".into(),
    })
    .await?;

use catalyst::generated::{
    CatalystActionSubjectMode, CatalystRuleFirePolicy, CatalystRuleMatchMode,
};
use catalyst::rules::{create, ActionSpec, ConditionSpec, CreateRule, TriggerSpec};
use serde_json::json;

// The rule starts as a Draft. Backtest it, then go live.
let rule_id = create(
    &valence,
    CreateRule {
        organization_id: "org-1".into(),
        name: "Trim winners".into(),
        created_by: "user-a".into(),
        timezone: "America/Los_Angeles".into(),
        tag_ids: Vec::new(),
        subject_selector_json: json!({"mode": "One", "ids": ["acct-1"]}),
        match_mode: CatalystRuleMatchMode::All,
        fire_policy: CatalystRuleFirePolicy::Everytime,
        cooldown_secs: None,
        max_fires_per_day: Some(3),
        trigger: TriggerSpec {
            kind_id: "chronon.schedule".into(),
            params: json!({"freq": "monthly"}),
        },
        conditions: vec![ConditionSpec {
            kind_id: "finance.balance_gte".into(),
            params: json!({"account_id": "acct-1", "amount": 100}),
            sort: 0,
            group: None,
            group_match_mode: None,
        }],
        actions: vec![ActionSpec {
            kind_id: "finance.transfer.initiate".into(),
            params: json!({"source_account_id": "acct-1", "dest_account_id": "acct-2"}),
            sort: 0,
            subject_mode: CatalystActionSubjectMode::Matched,
        }],
    },
)
.await?;

use polaron::graph::{self, AddReportingEdgeInput};
use polaron::org::{self, CreateOrgInput};
use polaron::teams::{self, AddMemberInput, CreateTeamInput};

let acme = org::create(CreateOrgInput { name: "Acme".into() }, &valence).await?;
let team = |name: &str| CreateTeamInput {
    org_id: acme.id.clone(),
    name: name.into(),
    description: None,
    parent_id: None,
    responsibilities: vec![],
};
let platform = teams::create(team("Platform"), &valence).await?;
let infra = teams::create(team("Infra"), &valence).await?;

// Infra reports to Platform. A reverse edge is rejected as a cycle.
graph::add_reporting_edge(
    AddReportingEdgeInput {
        child_team_id: infra.id.clone(),
        parent_team_id: platform.id.clone(),
    },
    &valence,
)
.await?;

teams::add_member(
    AddMemberInput {
        team_id: infra.id.clone(),
        user_id: "alice".into(),
        seat_kind: None,
        position_id: None,
    },
    &valence,
)
.await?;

use magnon::goal::{self, CreateGoalInput, GoalableSource};
use magnon::metric::{self, CreateMetricInput, MetricTarget, ReadingInput, ReadingSourceKind};

let goal = goal::create(
    CreateGoalInput {
        source: GoalableSource::Org(hub.id.clone()),
        title: "Cut p99 latency 40%".into(),
        description: None,
        success_criteria: None,
        start_date: None,
        target_date: None,
    },
    &valence,
)
.await?;

let p99 = metric::create(
    CreateMetricInput {
        goal_id: goal.id.clone(),
        name: "p99 latency".into(),
        unit: "ms".into(),
        target: MetricTarget::AtMost { value: 120.0 },
        cadence: None,
    },
    &valence,
)
.await?;

metric::record_reading(
    &p99.id,
    ReadingInput {
        value: 118.0,
        recorded_at: None,
        note: None,
        source_kind: ReadingSourceKind::Manual,
        external_ref: None,
    },
    &valence,
)
.await?;

let progress = goal::goal_progress(&goal.id, &valence).await?;
assert_eq!(progress.state.pct(), Some(100));