honey-biscuit-workshop/workshop-baker/src/main.rs

use workshop_baker::types::resource::ResourceRegistry;
use std::path::PathBuf;

use workshop_baker::{
    bake::bake, cli::{Cli, Commands, Parser},
    error::CliError,
    finalize::finalize, notify::notify, prebake::prebake,
};
use workshop_baker::bare::run_bare;
use workshop_baker::notify::queue::NotificationQueue;
use workshop_engine::{EventSender, ExecutionContext};

fn or_workshop(path: &Option<PathBuf>, name: &str) -> PathBuf {
    path.clone().unwrap_or_else(|| PathBuf::from(format!(".workshop/{}", name)))
}

async fn worker_main(cli: Cli) -> Result<(), CliError> {
    let prebake_path = cli.prebake.clone().unwrap_or_else(|| {
        log::error!("--prebake is required in worker standalone mode");
        std::process::exit(1);
    });
    let bake_path = cli.bake.clone().unwrap_or_else(|| {
        log::error!("--bake is required in worker standalone mode");
        std::process::exit(1);
    });
    let finalize_path = cli.finalize.clone().unwrap_or_else(|| {
        log::error!("--finalize is required in worker standalone mode");
        std::process::exit(1);
    });

    log::info!("Worker standalone mode");

    let mut ctx = ExecutionContext {
        standalone: true,
        task_id: format!("{}-{}-worker", cli.pipeline, cli.build_id),
        pipeline_name: cli.pipeline.clone(),
        username: cli.username.clone(),
        dry_run: cli.dry_run,
        privileged: false,
        ..Default::default()
    };

    let (tx, event_rx) = tokio::sync::mpsc::channel(256);
    let event_tx: EventSender = Some(tx);
    let _event_handle = tokio::spawn(workshop_baker::prebake::event::event_receiver(
        event_rx, cli.pipeline.clone(), cli.build_id.clone(),
    ));

    let (to_tx, to_rx) = tokio::sync::mpsc::channel(256);
    let (retry_tx, retry_rx) = tokio::sync::mpsc::channel(256);

    // Resource registry — populated before stages, consumed by stages
    let registry = ResourceRegistry::new();
    // TODO: pre-fetch plugins and templates into registry

    let notify_path = finalize_path.clone();
    let notify_ctx = ctx.clone();
    let notify_etx = event_tx.clone();
    let debug = cli.debug.clone();
    let notify_registry = registry.clone();
    let notify_handle = tokio::spawn(async move {
        notify(&notify_path, &mut notify_ctx.clone(), notify_etx, to_rx, retry_tx, debug.map(|d| d.contains("notify")).unwrap_or(false), &notify_registry).await
    });

    let nevent_tx = to_tx.clone();
    let mut queue = NotificationQueue::new(to_tx, retry_rx);

    let pipe_result: anyhow::Result<()> = async {
        prebake(&prebake_path, &cli, None, &mut ctx, event_tx.clone(), Some(nevent_tx)).await?;
        let bake_base_path = or_workshop(&cli.bake_base, "bake_base.sh");
        let pb_path = or_workshop(&cli.prebake, "prebake.yml");
        bake(&bake_path, &bake_base_path, &pb_path, &cli, &mut ctx, event_tx.clone()).await?;
        finalize(&finalize_path, &cli, &mut ctx, event_tx.clone(), &registry).await?;
        Ok(())
    }.await;

    if let Err(e) = pipe_result {
        log::error!("Pipeline stage failed: {}", e);
        return Err(CliError::General(e));
    }

    log::info!("Pipeline complete. Draining notifications...");
    if let Err(e) = queue.drain().await {
        log::error!("Notification queue error: {}", e);
    }
    let _ = notify_handle.await;
    Ok(())
}

#[tokio::main]
async fn main() {
    env_logger::init();
    let cli = Cli::parse();
    if cli.dry_run {
        log::warn!("Dry run enabled, no changes will be made.");
    }

    let pipeline = cli.pipeline.clone();
    let build_id = cli.build_id.clone();

    match &cli.command {
        Some(Commands::Bare { command }) => {
            let mut ctx = ExecutionContext {
                standalone: true,
                task_id: format!("{}-{}-bare", pipeline, build_id),
                pipeline_name: pipeline.clone(),
                username: cli.username.clone(),
                dry_run: cli.dry_run,
                privileged: false,
                ..Default::default()
            };
            let (tx, event_rx) = tokio::sync::mpsc::channel(256);
            let event_tx: EventSender = Some(tx);
            let event_handle = tokio::spawn(workshop_baker::prebake::event::event_receiver(
                event_rx, pipeline.clone(), build_id.clone(),
            ));

            let registry = ResourceRegistry::new();
            let result = run_bare(&cli, command, &mut ctx, event_tx, &pipeline, &build_id, &registry).await;

            drop(ctx);
            let _ = event_handle.await;

            if let Err(e) = result {
                log::error!("{}", e);
                std::process::exit(e.exit_code());
            }
        }
        Some(Commands::Daemon {}) => {
            unimplemented!("Daemon mode not implemented");
        }
        None => {
            if let Err(e) = worker_main(cli).await {
                log::error!("{}", e);
                std::process::exit(e.exit_code());
            }
        }
    }
}