Commit

Author:

Hash:

Timestamp:

+789 -21 +/-27 browse

Kevin Schoon [me@kevinschoon.com]

5defd227ab818cf7d0304670d0b685e7a6700e87

Fri, 03 Jul 2026 22:03:23 +0000 (1 week ago)

init ayllu-dispatch event processing service 
init ayllu-dispatch event processing service

This adds a new binary called `ayllu-dispatch` which is used to respond
to different effects across the forge. Currently dispatch integrates
only with a ntfy[1] based server however any UnifiedPush[2]
implementations will be supported in the future.

A subsequent change will implement event hooks for server-side git
operations which will be used to invoke builds automatically.

1. https://ntfy.sh
2. https://unifiedpush.org
1diff --git a/.ayllu-build.jsonnet b/.ayllu-build.jsonnet
2index 858822d..1193155 100644
3--- a/.ayllu-build.jsonnet
4+++ b/.ayllu-build.jsonnet
5 @@ -18,6 +18,7 @@ local build_cache = {
6
7 local binaries = [
8 'ayllu-build',
9+ 'ayllu-dispatch',
10 'ayllu-keys',
11 'ayllu-migrate',
12 'ayllu-shell',
13 diff --git a/Cargo.lock b/Cargo.lock
14index 44a7271..9ead26e 100644
15--- a/Cargo.lock
16+++ b/Cargo.lock
17 @@ -325,6 +325,24 @@ dependencies = [
18 ]
19
20 [[package]]
21+ name = "ayllu-dispatch"
22+ version = "0.1.0"
23+ dependencies = [
24+ "ayllu_cmd",
25+ "ayllu_config",
26+ "ayllu_database",
27+ "ayllu_git",
28+ "ayllu_identity",
29+ "ayllu_logging",
30+ "derive_builder",
31+ "reqwest",
32+ "serde",
33+ "thiserror 2.0.18",
34+ "tracing",
35+ "url",
36+ ]
37+
38+ [[package]]
39 name = "ayllu-keys"
40 version = "0.5.0"
41 dependencies = [
42 @@ -442,6 +460,7 @@ dependencies = [
43 "tempfile",
44 "thiserror 2.0.18",
45 "toml 0.8.23",
46+ "url",
47 ]
48
49 [[package]]
50 @@ -2618,6 +2637,7 @@ checksum = "62e0021ea2c22aed41653bc7e1419abb2c97e038ff2c33d0e1309e49a97deec0"
51 dependencies = [
52 "base64 0.22.1",
53 "bytes",
54+ "futures-channel",
55 "futures-core",
56 "futures-util",
57 "h2",
58 diff --git a/Cargo.toml b/Cargo.toml
59index 3af70e5..70e72a0 100644
60--- a/Cargo.toml
61+++ b/Cargo.toml
62 @@ -9,14 +9,14 @@ members = [
63 "crates/identity",
64 "crates/timeutil",
65 "crates/database",
66- "ayllu-web",
67 "ayllu-build",
68- # "ayllu-mail",
69- "ayllu-shell",
70+ "ayllu-dispatch",
71 "ayllu-keys",
72- "ayllu-migrate",
73+ "ayllu-migrate",
74+ "ayllu-shell",
75+ "ayllu-web",
76 "quipu",
77- "xtask",
78+ "xtask",
79 ]
80
81 [workspace.dependencies]
82 @@ -40,7 +40,7 @@ tracing-subscriber = { version = "0.3.22", features = ["env-filter"] }
83 openssh-keys = "0.6.5"
84 url = { version = "2.5.8", features = ["serde"]}
85
86- reqwest = { version = "0.13.3", default-features = false, features = ["rustls", "json", "http2", "stream"] }
87+ reqwest = { version = "0.13.3", default-features = false, features = ["rustls", "json", "http2", "stream", "blocking"] }
88
89 tar = "0.4.45"
90 tokio = { version = "1.49.0", features = ["full"] }
91 diff --git a/PKGBUILD b/PKGBUILD
92index 11ebe40..73f6125 100644
93--- a/PKGBUILD
94+++ b/PKGBUILD
95 @@ -22,6 +22,7 @@ source=()
96
97 BINARIES=(
98 'ayllu-build'
99+ 'ayllu-dispatch'
100 'ayllu-keys'
101 'ayllu-migrate'
102 'ayllu-shell'
103 diff --git a/ayllu-dispatch/Cargo.toml b/ayllu-dispatch/Cargo.toml
104new file mode 100644
105index 0000000..34544f9
106--- /dev/null
107+++ b/ayllu-dispatch/Cargo.toml
108 @@ -0,0 +1,19 @@
109+ [package]
110+ name = "ayllu-dispatch"
111+ version = "0.1.0"
112+ edition = "2024"
113+
114+ [dependencies]
115+ ayllu_cmd = { path = "../crates/cmd" }
116+ ayllu_config = { path = "../crates/config" }
117+ ayllu_git = { path = "../crates/git"}
118+ ayllu_identity = { path = "../crates/identity" }
119+ ayllu_database = { path = "../crates/database" }
120+ ayllu_logging = { path = "../crates/logging" }
121+
122+ reqwest = { workspace = true }
123+ tracing = { workspace = true }
124+ serde = { workspace = true }
125+ thiserror = { workspace = true }
126+ derive_builder = { workspace = true }
127+ url = { workspace = true }
128 diff --git a/ayllu-dispatch/src/config.rs b/ayllu-dispatch/src/config.rs
129new file mode 100644
130index 0000000..ef4f5c2
131--- /dev/null
132+++ b/ayllu-dispatch/src/config.rs
133 @@ -0,0 +1,73 @@
134+ use ayllu_config::Configurable;
135+ use ayllu_database::events::Kind;
136+ use serde::{Deserialize, Serialize};
137+
138+ use crate::notify::Notifier;
139+
140+ #[derive(Serialize, Deserialize)]
141+ pub enum Actor {
142+ #[serde(rename = "notify")]
143+ Notify(crate::notify::Config),
144+ }
145+
146+ #[derive(Serialize, Deserialize)]
147+ pub struct Event {
148+ pub selectors: Vec<Kind>,
149+ pub actor: Actor,
150+ }
151+
152+ #[derive(Serialize, Deserialize)]
153+ pub struct Dispatch {
154+ #[serde(default = "Dispatch::default_timeout_secs")]
155+ pub timeout_secs: u64,
156+ #[serde(default = "Dispatch::default_poll_interval_ms")]
157+ pub poll_interval_ms: u64,
158+ pub notifier: Option<Notifier>,
159+ #[serde(default = "Vec::default")]
160+ pub on_event: Vec<Event>,
161+ }
162+
163+ impl Default for Dispatch {
164+ fn default() -> Self {
165+ Self {
166+ timeout_secs: Dispatch::default_timeout_secs(),
167+ poll_interval_ms: Dispatch::default_poll_interval_ms(),
168+ notifier: None,
169+ on_event: Vec::default(),
170+ }
171+ }
172+ }
173+
174+ impl Dispatch {
175+ fn default_timeout_secs() -> u64 {
176+ 3600
177+ }
178+
179+ fn default_poll_interval_ms() -> u64 {
180+ 800
181+ }
182+ }
183+
184+ #[derive(Serialize, Deserialize, Default)]
185+ pub struct Config {
186+ #[serde(flatten)]
187+ pub common: ayllu_config::Common,
188+ #[serde(default = "Dispatch::default")]
189+ pub dispatch: Dispatch,
190+ }
191+
192+ impl Configurable for Config {}
193+
194+ #[cfg(test)]
195+ mod tests {
196+ use super::*;
197+ use ayllu_config::Reader;
198+
199+ const EXAMPLE_CONFIG: &str = include_str!("../../config.example.toml");
200+
201+ #[test]
202+ fn test_example_config() {
203+ Reader::<Config>::read(EXAMPLE_CONFIG).unwrap();
204+ Reader::<Config>::read("").unwrap();
205+ }
206+ }
207 diff --git a/ayllu-dispatch/src/error.rs b/ayllu-dispatch/src/error.rs
208new file mode 100644
209index 0000000..8ac863e
210--- /dev/null
211+++ b/ayllu-dispatch/src/error.rs
212 @@ -0,0 +1,11 @@
213+ use reqwest::StatusCode;
214+
215+ #[derive(Debug, thiserror::Error)]
216+ pub enum Error {
217+ #[error("Config: {0}")]
218+ Config(#[from] ayllu_config::Error),
219+ #[error("Database: {0}")]
220+ Database(#[from] ayllu_database::Error),
221+ #[error("Failed to send notification: {status_code:?}")]
222+ NotificationFailed { status_code: Option<StatusCode> },
223+ }
224 diff --git a/ayllu-dispatch/src/main.rs b/ayllu-dispatch/src/main.rs
225new file mode 100644
226index 0000000..fff3ad2
227--- /dev/null
228+++ b/ayllu-dispatch/src/main.rs
229 @@ -0,0 +1,57 @@
230+ use std::time::{Duration, Instant};
231+
232+ use ayllu_cmd::dispatch::Command;
233+ use tracing::Level;
234+
235+ use crate::error::Error;
236+
237+ mod config;
238+ mod error;
239+ mod notify;
240+
241+ fn main() -> Result<(), Error> {
242+ let args = ayllu_cmd::parse::<Command>();
243+ let config: config::Config = ayllu_config::Reader::load(args.config.as_deref())?;
244+ ayllu_logging::init(Level::INFO);
245+ let mut db = ayllu_database::Wrapper::new(&config.common.database.path)?;
246+ let mut conn = db.call();
247+ let mut start_time = Instant::now();
248+ let timeout = Duration::from_secs(config.dispatch.timeout_secs);
249+ loop {
250+ while let Some(event) = conn.event_next()? {
251+ tracing::info!("Processing event: {event:?}");
252+ config
253+ .dispatch
254+ .on_event
255+ .iter()
256+ .try_for_each(|event_config| {
257+ match &event_config.actor {
258+ config::Actor::Notify(notify_cfg) => {
259+ if event_config.selectors.contains(&event.kind) {
260+ let notifier = config.dispatch.notifier.as_ref().unwrap();
261+ tracing::info!("Dispatching notification event");
262+ // TODO: Need timeout here
263+ notifier.on_event(
264+ &config.common.origin,
265+ &mut conn,
266+ &event,
267+ notify_cfg,
268+ )?;
269+ };
270+ }
271+ };
272+ Ok::<_, Error>(())
273+ })?;
274+ conn.event_acknowledge(event.id)?;
275+ start_time = Instant::now();
276+ }
277+ if start_time.elapsed() >= timeout {
278+ tracing::info!(
279+ "Shutting down dispatch service since no events have been processed in {timeout:?}"
280+ );
281+ break;
282+ }
283+ std::thread::sleep(Duration::from_millis(config.dispatch.poll_interval_ms));
284+ }
285+ Ok(())
286+ }
287 diff --git a/ayllu-dispatch/src/notify.rs b/ayllu-dispatch/src/notify.rs
288new file mode 100644
289index 0000000..ea03bb8
290--- /dev/null
291+++ b/ayllu-dispatch/src/notify.rs
292 @@ -0,0 +1,121 @@
293+ use ayllu_database::{Ptr, build::State, events::Kind};
294+ use serde::{Deserialize, Serialize};
295+ use url::Url;
296+
297+ use crate::error::Error;
298+
299+ #[derive(Serialize, Deserialize)]
300+ pub struct Config {
301+ pub topic: String,
302+ }
303+
304+ /// Notifier implements server <--> distributor support for a
305+ /// https://unifiedpush.org compatible server.
306+ #[derive(Serialize, Deserialize)]
307+ pub enum Notifier {
308+ #[serde(rename = "ntfy")]
309+ Ntfy { endpoint: Url },
310+ }
311+
312+ impl Notifier {
313+ pub fn on_event(
314+ &self,
315+ base_url: &Url,
316+ db: &mut Ptr<'_>,
317+ event: &ayllu_database::events::Event,
318+ config: &Config,
319+ ) -> Result<(), Error> {
320+ let client = reqwest::blocking::ClientBuilder::default()
321+ .user_agent(ayllu_config::AYLLU_USER_AGENT)
322+ .build()
323+ .unwrap();
324+ let mut req = match self {
325+ Notifier::Ntfy { endpoint } => {
326+ let mut url = endpoint.clone();
327+ url.set_path(config.topic.as_str());
328+ client.post(url)
329+ }
330+ };
331+ let object_id = event.object_id.unwrap_or_default();
332+ match event.kind {
333+ Kind::BuildStarted | Kind::BuildFinished => {
334+ let manifest = db.manifest_read(event.object_id.unwrap())?;
335+ let mut build_url = base_url.clone();
336+ build_url.set_path(&format!("/builds/{object_id}"));
337+ req = req.header("Actions", format!("view, View, {build_url}"));
338+ req = req.header(
339+ "Tags",
340+ match manifest.state {
341+ State::Passed => "white_check_mark",
342+ State::Failed => "warning",
343+ _ => "information_source",
344+ },
345+ );
346+ let build_key = match (manifest.collection, manifest.name) {
347+ (Some(collection), Some(name)) => format!("{collection}/{name}"),
348+ _ => "Anonymous".to_string(),
349+ };
350+ req = req.body(format!(
351+ "Build {object_id}: {build_key} has entered state: {}",
352+ manifest.state
353+ ));
354+ }
355+ Kind::WorkflowStarted | Kind::WorkflowFinished => {
356+ let workflow = db.workflow_read(event.object_id.unwrap())?;
357+ let manifest = db.manifest_read(workflow.manifest_id)?;
358+ let mut workflow_url = base_url.clone();
359+ workflow_url.set_path(&format!("/builds/{}/{object_id}", manifest.id));
360+ req = req.header("Actions", format!("view, View, {workflow_url}"));
361+ req = req.header(
362+ "Tags",
363+ match workflow.state {
364+ State::Passed => "white_check_mark",
365+ State::Failed => "warning",
366+ _ => "information_source",
367+ },
368+ );
369+ let build_key = match (manifest.collection, manifest.name) {
370+ (Some(collection), Some(name)) => format!("{collection}/{name}"),
371+ _ => "Anonymous".to_string(),
372+ };
373+ req = req.body(format!(
374+ "Workflow {object_id}: {build_key} has entered state: {}",
375+ manifest.state
376+ ));
377+ }
378+ Kind::StepStarted | Kind::StepFinished => {
379+ let step = db.step_read(object_id)?;
380+ let workflow = db.workflow_read(step.workflow_id)?;
381+ let manifest = db.manifest_read(workflow.manifest_id)?;
382+
383+ let mut step_url = base_url.clone();
384+ step_url.set_path(&format!(
385+ "/builds/{}/{object_id}/step/{}",
386+ workflow.id, step.id
387+ ));
388+ req = req.header("Actions", format!("view, View, {step_url}"));
389+ req = req.header(
390+ "Tags",
391+ match workflow.state {
392+ State::Passed => "white_check_mark",
393+ State::Failed => "warning",
394+ _ => "information_source",
395+ },
396+ );
397+ let build_key = match (manifest.collection, manifest.name) {
398+ (Some(collection), Some(name)) => format!("{collection}/{name}"),
399+ _ => "Anonymous".to_string(),
400+ };
401+ req = req.body(format!(
402+ "Step {object_id}: {build_key} has entered state: {}",
403+ manifest.state
404+ ));
405+ }
406+ _ => return Ok(()),
407+ }
408+ req.send().map_err(|e| Error::NotificationFailed {
409+ status_code: e.status(),
410+ })?;
411+ Ok(())
412+ }
413+ }
414 diff --git a/ayllu-dispatch/src/process.rs b/ayllu-dispatch/src/process.rs
415new file mode 100644
416index 0000000..3f53794
417--- /dev/null
418+++ b/ayllu-dispatch/src/process.rs
419 @@ -0,0 +1,75 @@
420+ use ayllu_cmd::init::Command;
421+ use nix::sys::signal::{self, SigSet, Signal};
422+ use std::{
423+ fs::OpenOptions,
424+ io::{BufRead, BufReader, Read, Write},
425+ path::Path,
426+ process::{Child, Stdio},
427+ sync::mpsc::{self, channel, Receiver, Sender},
428+ thread::JoinHandle,
429+ time::{Duration, SystemTime},
430+ };
431+
432+ fn trap(shutdown: Option<Sender<Sender<()>>>) {
433+ let mask = SigSet::all();
434+ signal::pthread_sigmask(signal::SigmaskHow::SIG_BLOCK, Some(&mask), None).unwrap();
435+ loop {
436+ tracing::info!("Waiting for signal");
437+ let sig = mask.wait().unwrap();
438+ tracing::info!("Caught signal {sig:?}");
439+ match sig {
440+ Signal::SIGINT => break,
441+ Signal::SIGQUIT => break,
442+ Signal::SIGTERM => break,
443+ Signal::SIGCHLD => {
444+ tracing::info!("Child process exited");
445+ return;
446+ }
447+ _ => {}
448+ }
449+ }
450+ if let Some(shutdown) = shutdown {
451+ let (tx, rx) = mpsc::channel::<()>();
452+ shutdown.send(tx).unwrap();
453+ rx.recv_timeout(Duration::from_secs(10)).unwrap();
454+ }
455+ }
456+
457+ // fn main() -> Result<(), Box<dyn std::error::Error>> {
458+ // let args = ayllu_cmd::parse::<Command>();
459+ // ayllu_logging::init(ayllu_logging::Level::INFO);
460+ // if !args.force && !within_ayllu_build_container() {
461+ // return Err("ayllu-init can only be run from within a build container".into());
462+ // };
463+ // match args.command {
464+ // ayllu_cmd::init::Subcommand::Init => {
465+ // initialize()?;
466+ // trap(None);
467+ // Ok(())
468+ // }
469+ // ayllu_cmd::init::Subcommand::Exec {
470+ // check,
471+ // forward,
472+ // args,
473+ // step,
474+ // } => {
475+ // if check {
476+ // tracing::info!("Waiting for build environment to be ready");
477+ // ready()?;
478+ // tracing::info!("Environment is ready!");
479+ // };
480+ // let exec_args = args.clone();
481+ // let (_, shutdown) = exec(
482+ // exec_args.as_slice(),
483+ // step,
484+ // forward,
485+ // Path::new(&format!("/tmp/ayllu.{}.stdout", step)),
486+ // Path::new(&format!("/tmp/ayllu.{}.stderr", step)),
487+ // Path::new(&format!("/tmp/ayllu.{}.exit_code", step)),
488+ // Duration::from_secs(MAX_RUNTIME_SECONDS),
489+ // )?;
490+ // trap(Some(shutdown));
491+ // Ok(())
492+ // }
493+ // }
494+ // }
495 diff --git a/config.example.toml b/config.example.toml
496index 83e13fe..db338bb 100644
497--- a/config.example.toml
498+++ b/config.example.toml
499 @@ -263,3 +263,18 @@ args = ["-"]
500 [[quipu.instances]]
501 name = "ayllu-forge"
502 url = "https://ayllu-forge.org"
503+
504+
505+ ###
506+ #### Dispatch Configuration
507+ ###
508+
509+ [dispatch]
510+ [dispatch.notifier.ntfy]
511+ endpoint = "http://127.0.0.1:8000"
512+
513+ [[dispatch.on_event]]
514+ selectors = ["BuildStarted", "BuildFinished"]
515+ # NOTE that a wildcard here will result in MANY notifications
516+ # selectors = ["*"]
517+ actor = { notify = { topic = "ayllu-main" } }
518 diff --git a/crates/cmd/src/dispatch.rs b/crates/cmd/src/dispatch.rs
519new file mode 100644
520index 0000000..a2b0a92
521--- /dev/null
522+++ b/crates/cmd/src/dispatch.rs
523 @@ -0,0 +1,30 @@
524+ use std::path::PathBuf;
525+
526+ use clap::Parser;
527+ // use clap_complete::Shell;
528+ use tracing::Level;
529+
530+ /// Ayllu event dispatching service
531+ #[derive(Parser)]
532+ #[command(
533+ author,
534+ version,
535+ about,
536+ name = "quipu",
537+ long_about = r#"
538+
539+ ayllu-dispatch listens for events to occur and performs some action depending on
540+ the type of event being processed. To inspect pending events manually you can
541+ use the quipu(1) events command. Once the process starts up it will process
542+ all pending events and then periodically poll for new events until the timeout
543+ is reached and then shutdown.
544+ "#
545+ )]
546+ pub struct Command {
547+ /// Path to your configuration file
548+ #[arg(short, long, value_name = "FILE")]
549+ pub config: Option<PathBuf>,
550+ /// Sets the logging level
551+ #[arg(short, long, value_name = "LEVEL")]
552+ pub level: Option<Level>,
553+ }
554 diff --git a/crates/cmd/src/lib.rs b/crates/cmd/src/lib.rs
555index 7194359..6851455 100644
556--- a/crates/cmd/src/lib.rs
557+++ b/crates/cmd/src/lib.rs
558 @@ -1,5 +1,6 @@
559 pub mod ayllu;
560 pub mod build;
561+ pub mod dispatch;
562 pub mod keys;
563 pub mod migrate;
564 pub mod quipu;
565 diff --git a/crates/cmd/src/quipu.rs b/crates/cmd/src/quipu.rs
566index 72419c5..87b22e2 100644
567--- a/crates/cmd/src/quipu.rs
568+++ b/crates/cmd/src/quipu.rs
569 @@ -103,4 +103,6 @@ pub enum Subcommand {
570 #[command(subcommand)]
571 build: Build,
572 },
573+ /// Firehose of events in the forge
574+ Events,
575 }
576 diff --git a/crates/config/Cargo.toml b/crates/config/Cargo.toml
577index fb65ca1..ca5b4d7 100644
578--- a/crates/config/Cargo.toml
579+++ b/crates/config/Cargo.toml
580 @@ -10,6 +10,8 @@ toml = { workspace = true }
581 serde = { workspace = true }
582 thiserror = {workspace = true }
583 clap = { workspace = true }
584+ url = { workspace = true }
585+
586 serde_json = "1.0.149"
587
588 [dev-dependencies]
589 diff --git a/crates/config/src/lib.rs b/crates/config/src/lib.rs
590index 2b962ad..d0432a5 100644
591--- a/crates/config/src/lib.rs
592+++ b/crates/config/src/lib.rs
593 @@ -1,4 +1,5 @@
594 use serde::{Deserialize, Serialize};
595+ use url::Url;
596
597 use std::path::{Path, PathBuf};
598
599 @@ -10,12 +11,15 @@ mod reader;
600
601 pub const DEFAULT_BRANCH_NAME: &str = "master";
602 pub const EXAMPLE_CONFIG: &str = include_str!("../../../config.example.toml");
603+ pub const AYLLU_USER_AGENT: &str = concat!(env!("CARGO_PKG_NAME"), "/", env!("CARGO_PKG_VERSION"));
604
605 /// Common configuration options across all binaries
606 #[derive(Deserialize, Serialize, Clone, Debug)]
607 pub struct Common {
608 #[serde(default = "Common::default_log_level")]
609 pub log_level: String,
610+ #[serde(default = "Common::default_origin")]
611+ pub origin: Url,
612 #[serde(default = "Common::default_branch")]
613 pub default_branch: String,
614 pub base_path: Option<PathBuf>,
615 @@ -32,6 +36,10 @@ impl Common {
616 fn default_branch() -> String {
617 DEFAULT_BRANCH_NAME.to_string()
618 }
619+
620+ fn default_origin() -> Url {
621+ Url::parse("http://127.0.0.1:10000").unwrap()
622+ }
623 }
624
625 impl Default for Common {
626 @@ -42,6 +50,7 @@ impl Default for Common {
627 database: Database::default(),
628 sysadmin: None,
629 base_path: None,
630+ origin: Common::default_origin(),
631 }
632 }
633 }
634 diff --git a/crates/database/migrations/2026-05-04-162320-0000_init/up.sql b/crates/database/migrations/2026-05-04-162320-0000_init/up.sql
635index 6ecd90f..7de3921 100644
636--- a/crates/database/migrations/2026-05-04-162320-0000_init/up.sql
637+++ b/crates/database/migrations/2026-05-04-162320-0000_init/up.sql
638 @@ -78,3 +78,11 @@ CREATE TABLE samples (
639 cpu_user_usec BIGINT NOT NULL,
640 cpu_system_usec BIGINT NOT NULL
641 ) ;
642+
643+ CREATE TABLE events (
644+ id INTEGER PRIMARY KEY NOT NULL,
645+ timestamp BIGINT NOT NULL,
646+ kind INTEGER NOT NULL,
647+ object_id INTEGER,
648+ acknowledged BOOLEAN NOT NULL DEFAULT false
649+ ) ;
650 diff --git a/crates/database/src/build.rs b/crates/database/src/build.rs
651index 008b047..b4c041d 100644
652--- a/crates/database/src/build.rs
653+++ b/crates/database/src/build.rs
654 @@ -330,6 +330,7 @@ pub mod manifests {
655 dsl::state.eq(State::Running),
656 ))
657 .execute(self.0)?;
658+ self.event_submit(crate::events::Kind::BuildStarted, Some(id))?;
659 Ok(())
660 }
661
662 @@ -341,6 +342,7 @@ pub mod manifests {
663 dsl::state.eq(state),
664 ))
665 .execute(self.0)?;
666+ self.event_submit(crate::events::Kind::BuildFinished, Some(id))?;
667 Ok(())
668 }
669 }
670 @@ -427,6 +429,7 @@ pub mod workflows {
671 dsl::state.eq(State::Running),
672 ))
673 .execute(self.0)?;
674+ self.event_submit(crate::events::Kind::WorkflowStarted, Some(workflow_id))?;
675 Ok(())
676 }
677
678 @@ -444,6 +447,7 @@ pub mod workflows {
679 dsl::duration.eq(duration),
680 ))
681 .execute(self.0)?;
682+ self.event_submit(crate::events::Kind::WorkflowFinished, Some(workflow_id))?;
683 Ok(())
684 }
685 }
686 @@ -517,6 +521,7 @@ pub mod steps {
687 dsl::state.eq(State::Running),
688 ))
689 .execute(self.0)?;
690+ self.event_submit(crate::events::Kind::StepStarted, Some(id))?;
691 Ok(())
692 }
693
694 @@ -536,6 +541,7 @@ pub mod steps {
695 dsl::duration.eq(duration),
696 ))
697 .execute(self.0)?;
698+ self.event_submit(crate::events::Kind::StepFinished, Some(id))?;
699 Ok(())
700 }
701 }
702 diff --git a/crates/database/src/events.rs b/crates/database/src/events.rs
703new file mode 100644
704index 0000000..879df0c
705--- /dev/null
706+++ b/crates/database/src/events.rs
707 @@ -0,0 +1,236 @@
708+ use std::time::{SystemTime, UNIX_EPOCH};
709+
710+ use diesel::{
711+ BoolExpressionMethods, ExpressionMethods, Identifiable, QueryDsl, Queryable, RunQueryDsl,
712+ Selectable, SelectableHelper,
713+ backend::Backend,
714+ deserialize::{self, FromSql, FromSqlRow},
715+ expression::AsExpression,
716+ serialize::{self, Output, ToSql},
717+ sql_types::Integer,
718+ sqlite::Sqlite,
719+ };
720+ use serde::{Deserialize, Serialize, de::Visitor};
721+ use tabled::Tabled;
722+
723+ use crate::{Error, Ptr, tabled_util};
724+
725+ use crate::schema::events::{dsl, table};
726+
727+ const ALL_KINDS: &[Kind] = &[
728+ Kind::Unspecified,
729+ Kind::BuildStarted,
730+ Kind::BuildFinished,
731+ Kind::WorkflowStarted,
732+ Kind::WorkflowFinished,
733+ Kind::StepStarted,
734+ Kind::StepFinished,
735+ ];
736+
737+ #[derive(AsExpression, Debug, Default, Clone, Copy, FromSqlRow)]
738+ #[diesel(sql_type = Integer)]
739+ pub enum Kind {
740+ #[default]
741+ Unspecified = 0,
742+ BuildStarted = 1,
743+ BuildFinished = 2,
744+ WorkflowStarted = 3,
745+ WorkflowFinished = 4,
746+ StepStarted = 5,
747+ StepFinished = 6,
748+ Wildcard = 255,
749+ }
750+
751+ impl PartialEq for Kind {
752+ fn eq(&self, other: &Self) -> bool {
753+ matches!(other, Self::Wildcard)
754+ || matches!(self, Self::Wildcard)
755+ || core::mem::discriminant(self) == core::mem::discriminant(other)
756+ }
757+ }
758+
759+ struct KindVisitor;
760+
761+ impl<'de> Visitor<'de> for KindVisitor {
762+ type Value = Kind;
763+
764+ fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result {
765+ formatter.write_str("Kind must be a string")
766+ }
767+
768+ fn visit_str<E>(self, v: &str) -> Result<Self::Value, E>
769+ where
770+ E: serde::de::Error,
771+ {
772+ match ALL_KINDS.iter().find(|kind| kind.to_string() == v) {
773+ Some(kind) => Ok(*kind),
774+ None => {
775+ if v == "*" {
776+ Ok(Kind::Wildcard)
777+ } else {
778+ Err(serde::de::Error::invalid_value(
779+ serde::de::Unexpected::Str(v),
780+ &self,
781+ ))
782+ }
783+ }
784+ }
785+ }
786+ }
787+
788+ impl Serialize for Kind {
789+ fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
790+ where
791+ S: serde::Serializer,
792+ {
793+ serializer.serialize_str(&self.to_string())
794+ }
795+ }
796+
797+ impl<'de> Deserialize<'de> for Kind {
798+ fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
799+ where
800+ D: serde::Deserializer<'de>,
801+ {
802+ deserializer.deserialize_str(KindVisitor)
803+ }
804+ }
805+
806+ impl std::fmt::Display for Kind {
807+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
808+ match self {
809+ Kind::Unspecified => write!(f, "Unspecified"),
810+ Kind::BuildStarted => write!(f, "BuildStarted"),
811+ Kind::BuildFinished => write!(f, "BuildFinished"),
812+ Kind::WorkflowStarted => write!(f, "WorkflowStarted"),
813+ Kind::WorkflowFinished => write!(f, "WorkflowFinished"),
814+ Kind::StepStarted => write!(f, "StepStarted"),
815+ Kind::StepFinished => write!(f, "StepFinished"),
816+ Kind::Wildcard => write!(f, "*"),
817+ }
818+ }
819+ }
820+
821+ impl<DB> FromSql<Integer, DB> for Kind
822+ where
823+ DB: Backend,
824+ i32: FromSql<Integer, DB>,
825+ {
826+ fn from_sql(bytes: DB::RawValue<'_>) -> deserialize::Result<Self> {
827+ match i32::from_sql(bytes)? {
828+ 0 => Ok(Kind::Unspecified),
829+ 1 => Ok(Kind::BuildStarted),
830+ 2 => Ok(Kind::BuildFinished),
831+ 3 => Ok(Kind::WorkflowStarted),
832+ 4 => Ok(Kind::WorkflowFinished),
833+ 5 => Ok(Kind::StepStarted),
834+ 6 => Ok(Kind::StepFinished),
835+ 255 => unreachable!(),
836+ x => Err(format!("Unrecognized variant {x}").into()),
837+ }
838+ }
839+ }
840+
841+ impl<DB> ToSql<Integer, DB> for Kind
842+ where
843+ DB: Backend,
844+ i32: ToSql<Integer, DB>,
845+ {
846+ fn to_sql<'b>(&'b self, out: &mut Output<'b, '_, DB>) -> serialize::Result {
847+ match self {
848+ Kind::Unspecified => 0_i32.to_sql(out),
849+ Kind::BuildStarted => 1_i32.to_sql(out),
850+ Kind::BuildFinished => 2_i32.to_sql(out),
851+ Kind::WorkflowStarted => 3_i32.to_sql(out),
852+ Kind::WorkflowFinished => 4_i32.to_sql(out),
853+ Kind::StepStarted => 5_i32.to_sql(out),
854+ Kind::StepFinished => 6_i32.to_sql(out),
855+ Kind::Wildcard => panic!("BUG: You cannot serialize the wildcard Kind"),
856+ }
857+ }
858+ }
859+
860+ /// Some event which is sometime related to a repository or to another object
861+ /// in the database.
862+ #[derive(
863+ Clone,
864+ Debug,
865+ Default,
866+ Deserialize,
867+ Serialize,
868+ Queryable,
869+ Selectable,
870+ PartialEq,
871+ Identifiable,
872+ Tabled,
873+ )]
874+ #[diesel(table_name = crate::schema::events)]
875+ #[diesel(check_for_backend(Sqlite))]
876+ pub struct Event {
877+ pub id: i32,
878+ #[tabled(display("tabled_util::i32_opt"))]
879+ pub object_id: Option<i32>,
880+ #[tabled(display("tabled_util::timestamp"))]
881+ pub timestamp: i64,
882+ pub kind: Kind,
883+ pub acknowledged: bool,
884+ }
885+
886+ impl Ptr<'_> {
887+ pub fn event_submit(&mut self, kind: Kind, object_id: Option<i32>) -> Result<i32, Error> {
888+ let timestamp = SystemTime::now();
889+ let timestamp = timestamp.duration_since(UNIX_EPOCH).unwrap();
890+ Ok(diesel::insert_into(table)
891+ .values((
892+ dsl::kind.eq(kind),
893+ dsl::object_id.eq(object_id),
894+ dsl::timestamp.eq(timestamp.as_secs() as i64),
895+ ))
896+ .returning(dsl::id)
897+ .get_result(self.0)?)
898+ }
899+
900+ /// return all events that occurred in the given timerange, if start is not
901+ /// specified it will return all events from the last 24 hours.
902+ pub fn event_range(
903+ &mut self,
904+ start: Option<i64>,
905+ end: Option<i64>,
906+ ) -> Result<Vec<Event>, Error> {
907+ let start = start.unwrap_or(
908+ (std::time::SystemTime::now()
909+ .duration_since(UNIX_EPOCH)
910+ .unwrap_or_default()
911+ .as_secs()
912+ - 86400) as i64,
913+ );
914+ let end = end.unwrap_or(
915+ std::time::SystemTime::now()
916+ .duration_since(UNIX_EPOCH)
917+ .unwrap_or_default()
918+ .as_secs() as i64,
919+ );
920+ Ok(table
921+ .select(Event::as_select())
922+ .filter(dsl::timestamp.gt(start).and(dsl::timestamp.lt(end)))
923+ .get_results(self.0)?)
924+ }
925+
926+ pub fn event_acknowledge(&mut self, id: i32) -> Result<(), Error> {
927+ diesel::update(table)
928+ .filter(dsl::id.eq(id))
929+ .set(dsl::acknowledged.eq(true))
930+ .execute(self.0)?;
931+ Ok(())
932+ }
933+
934+ pub fn event_next(&mut self) -> Result<Option<Event>, Error> {
935+ Ok(table
936+ .select(Event::as_select())
937+ .filter(dsl::acknowledged.eq(false))
938+ .limit(1)
939+ .get_results(self.0)?
940+ .first()
941+ .cloned())
942+ }
943+ }
944 diff --git a/crates/database/src/lib.rs b/crates/database/src/lib.rs
945index 2f0c383..8bc0cee 100644
946--- a/crates/database/src/lib.rs
947+++ b/crates/database/src/lib.rs
948 @@ -5,6 +5,7 @@ use diesel_migrations::{EmbeddedMigrations, MigrationHarness, embed_migrations};
949
950 pub mod build;
951 pub mod error;
952+ pub mod events;
953 pub mod schema;
954 pub mod tabled_util;
955
956 diff --git a/crates/database/src/schema.rs b/crates/database/src/schema.rs
957index 5f03a53..0b32ddb 100644
958--- a/crates/database/src/schema.rs
959+++ b/crates/database/src/schema.rs
960 @@ -12,6 +12,16 @@ diesel::table! {
961 }
962
963 diesel::table! {
964+ events (id) {
965+ id -> Integer,
966+ timestamp -> BigInt,
967+ kind -> Integer,
968+ object_id -> Nullable<Integer>,
969+ acknowledged -> Bool,
970+ }
971+ }
972+
973+ diesel::table! {
974 logs (id) {
975 id -> Integer,
976 step_id -> Integer,
977 @@ -105,5 +115,5 @@ diesel::joinable!(variables -> steps (step_id));
978 diesel::joinable!(workflows -> manifests (manifest_id));
979
980 diesel::allow_tables_to_appear_in_same_query!(
981- cache, logs, manifests, outputs, samples, steps, variables, workflows,
982+ cache, events, logs, manifests, outputs, samples, steps, variables, workflows,
983 );
984 diff --git a/crates/database/src/tabled_util.rs b/crates/database/src/tabled_util.rs
985index 4a21238..9ae80f6 100644
986--- a/crates/database/src/tabled_util.rs
987+++ b/crates/database/src/tabled_util.rs
988 @@ -18,6 +18,10 @@ pub fn friendly(input: &Option<u64>) -> String {
989 }
990 }
991
992+ pub fn timestamp(timestamp: &i64) -> String {
993+ timeutil::timestamp_from_epoch(*timestamp)
994+ }
995+
996 pub fn timestamp_opt(timestamp: &Option<i32>) -> String {
997 timestamp
998 .and_then(|ts| Some(timeutil::timestamp_from_epoch(ts as i64)))
999 diff --git a/quipu/src/main.rs b/quipu/src/main.rs
1000index c67dcf6..70af929 100644
1001--- a/quipu/src/main.rs
1002+++ b/quipu/src/main.rs
1003 @@ -9,12 +9,13 @@ use ayllu_cmd::quipu::{Command, Subcommand};
1004 use ayllu_config::Reader;
1005 use ayllu_database::{Blob, Wrapper as Database};
1006
1007- use crate::{error::QuipuError, ui::Ui};
1008+ use crate::{error::QuipuError, scope::Scope, ui::Ui};
1009
1010 mod client;
1011 mod config;
1012 mod error;
1013 mod output;
1014+ mod scope;
1015 mod ui;
1016
1017 fn get_instance(
1018 @@ -97,17 +98,11 @@ async fn main() -> Result<(), error::QuipuError> {
1019 let mut db = Database::new_ro(&cfg.database.path)?;
1020 let blob = Blob::new(&cfg.database.path)?;
1021 let mut conn = db.call();
1022- let current_path = std::env::current_dir()?.canonicalize()?;
1023- let (collection, name) = if !cli.global {
1024- let (collection, name) = ayllu_git::collection_and_name(current_path.as_path());
1025- (Some(collection), Some(name))
1026- } else {
1027- (None, None)
1028- };
1029+ let scope = Scope::new(cli.global)?;
1030 match build {
1031 ayllu_cmd::quipu::Build::List { limit, offset } => {
1032 let manifests =
1033- conn.manifest_list(collection.as_deref(), name.as_deref(), limit, offset)?;
1034+ conn.manifest_list(scope.collection(), scope.name(), limit, offset)?;
1035 Ui::Builds(manifests.as_slice()).display(cli.json);
1036 Ok(())
1037 }
1038 @@ -116,7 +111,7 @@ async fn main() -> Result<(), error::QuipuError> {
1039 id
1040 } else {
1041 if let Some(manifest_id) =
1042- conn.manifest_latest_id(collection.as_deref(), name.as_deref())?
1043+ conn.manifest_latest_id(scope.collection(), scope.name())?
1044 {
1045 manifest_id
1046 } else {
1047 @@ -136,7 +131,7 @@ async fn main() -> Result<(), error::QuipuError> {
1048 id
1049 } else {
1050 if let Some(manifest_id) =
1051- conn.manifest_latest_id(collection.as_deref(), name.as_deref())?
1052+ conn.manifest_latest_id(scope.collection(), scope.name())?
1053 {
1054 manifest_id
1055 } else {
1056 @@ -232,5 +227,27 @@ async fn main() -> Result<(), error::QuipuError> {
1057 },
1058 }
1059 }
1060+ Subcommand::Events => {
1061+ let mut db = Database::new_ro(&cfg.database.path)?;
1062+ let mut conn = db.call();
1063+ // let scope = Scope::new(cli.global)?;
1064+ let events = conn.event_range(None, None)?;
1065+ // TODO: scope based filtering not yet implemented
1066+ // let events = match scope {
1067+ // Scope::Repository { collection, name } => {
1068+ // events.iter().try_fold(Vec::new(), |mut accm, event| {
1069+ // match event.kind {
1070+ // Kind::BuildStarted | Kind::BuildFinished => todo!(),
1071+ // Kind::WorkflowStarted | Kind::WorkflowFinished => todo!(),
1072+ // Kind::StepStarted | Kind::StepFinished => todo!(),
1073+ // _ => {}
1074+ // }
1075+ // })?
1076+ // },
1077+ // Scope::Global => events,
1078+ // };
1079+ Ui::Events(events.as_slice()).display(cli.json);
1080+ Ok(())
1081+ }
1082 }
1083 }
1084 diff --git a/quipu/src/scope.rs b/quipu/src/scope.rs
1085new file mode 100644
1086index 0000000..f5f14a6
1087--- /dev/null
1088+++ b/quipu/src/scope.rs
1089 @@ -0,0 +1,32 @@
1090+ use crate::error::QuipuError;
1091+
1092+ pub enum Scope {
1093+ Repository { collection: String, name: String },
1094+ Global,
1095+ }
1096+
1097+ impl Scope {
1098+ pub fn new(global: bool) -> Result<Self, QuipuError> {
1099+ if global {
1100+ Ok(Self::Global)
1101+ } else {
1102+ let current_path = std::env::current_dir()?.canonicalize()?;
1103+ let (collection, name) = ayllu_git::collection_and_name(current_path.as_path());
1104+ Ok(Self::Repository { collection, name })
1105+ }
1106+ }
1107+
1108+ pub fn collection(&self) -> Option<&str> {
1109+ match self {
1110+ Scope::Repository { collection, .. } => Some(collection),
1111+ Scope::Global => None,
1112+ }
1113+ }
1114+
1115+ pub fn name(&self) -> Option<&str> {
1116+ match self {
1117+ Scope::Repository { name, .. } => Some(name),
1118+ Scope::Global => None,
1119+ }
1120+ }
1121+ }
1122 diff --git a/quipu/src/ui.rs b/quipu/src/ui.rs
1123index 061e0f0..cb1f48c 100644
1124--- a/quipu/src/ui.rs
1125+++ b/quipu/src/ui.rs
1126 @@ -1,6 +1,9 @@
1127- use ayllu_database::build::{
1128- manifests::{Manifest, ResolvedManifest},
1129- steps::Step,
1130+ use ayllu_database::{
1131+ build::{
1132+ manifests::{Manifest, ResolvedManifest},
1133+ steps::Step,
1134+ },
1135+ events::Event,
1136 };
1137
1138 use tabled::{Table, settings::Style};
1139 @@ -8,6 +11,7 @@ use tabled::{Table, settings::Style};
1140 pub enum Ui<'a> {
1141 Builds(&'a [Manifest]),
1142 Build(&'a ResolvedManifest),
1143+ Events(&'a [Event]),
1144 }
1145
1146 impl Ui<'_> {
1147 @@ -48,6 +52,13 @@ impl Ui<'_> {
1148 println!("{table}");
1149 }
1150 }
1151+ Ui::Events(events) => {
1152+ if as_json {
1153+ serde_json::to_writer(std::io::stdout(), events).unwrap()
1154+ } else {
1155+ println!("{}", Table::new(*events).with(Style::modern()));
1156+ }
1157+ }
1158 }
1159 }
1160 }
1161 diff --git a/scripts/build_deb.sh b/scripts/build_deb.sh
1162index 2e20d41..07530cb 100755
1163--- a/scripts/build_deb.sh
1164+++ b/scripts/build_deb.sh
1165 @@ -20,6 +20,7 @@ sed -i "s/Version: X.X.X/Version: ${AYLLU_VERSION}/g" "${BUILD_DIR}/DEBIAN/contr
1166
1167 BINARIES=(
1168 'ayllu-build'
1169+ 'ayllu-dispatch'
1170 'ayllu-keys'
1171 'ayllu-migrate'
1172 'ayllu-shell'
1173 diff --git a/xtask/src/main.rs b/xtask/src/main.rs
1174index e09234d..61a3da1 100644
1175--- a/xtask/src/main.rs
1176+++ b/xtask/src/main.rs
1177 @@ -117,6 +117,10 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
1178 &target_dir(DIST_COMPLETE_PATH),
1179 "ayllu-shell",
1180 )?;
1181+ complete_gen::<ayllu_cmd::dispatch::Command>(
1182+ &target_dir(DIST_COMPLETE_PATH),
1183+ "ayllu-dispatch",
1184+ )?;
1185 complete_gen::<ayllu_cmd::keys::Command>(
1186 &target_dir(DIST_COMPLETE_PATH),
1187 "ayllu-keys",
1188 @@ -139,6 +143,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
1189 std::fs::create_dir_all(&man_dir)?;
1190 // Binary man pages
1191 man_gen::<ayllu_cmd::ayllu::Command>(&man_dir)?;
1192+ man_gen::<ayllu_cmd::dispatch::Command>(&man_dir)?;
1193 man_gen::<ayllu_cmd::shell::Command>(&man_dir)?;
1194 man_gen::<ayllu_cmd::keys::Command>(&man_dir)?;
1195 man_gen::<ayllu_cmd::migrate::Command>(&man_dir)?;