use std::path::Path;
use std::time::{SystemTime, UNIX_EPOCH};
use crate::filter::mask_secrets;
use kindling_provider::{retrieve_at, LocalFtsProvider};
use kindling_store::{SqliteKindlingStore, StoreError, StoreOptions};
use kindling_types::{
Capsule, CapsuleInput, CapsuleStatus, CapsuleType, Id, ListObservationsRequest,
ListObservationsResult, Observation, ObservationInput, Pin, PinInput, PinTargetType,
RetrieveOptions, RetrieveResult, ScopeIds, Summary, SummaryInput, Timestamp, ValidationError,
};
use crate::context::{PreCompactContext, ResolvedPin, SessionStartContext};
use crate::error::{ServiceError, ServiceResult};
use crate::validation;
#[derive(Debug, Clone)]
pub struct OpenCapsuleOptions {
pub kind: CapsuleType,
pub intent: String,
pub scope_ids: ScopeIds,
pub id: Option<Id>,
}
#[derive(Debug, Clone, Default)]
pub struct CloseCapsuleOptions {
pub generate_summary: bool,
pub summary_content: Option<String>,
pub confidence: Option<f64>,
}
#[derive(Debug, Clone)]
pub struct AppendObservationOptions {
pub capsule_id: Option<Id>,
pub validate: bool,
}
#[derive(Debug, Clone, PartialEq)]
pub struct AppendOutcome {
pub observation: Observation,
pub deduplicated: bool,
}
impl Default for AppendObservationOptions {
fn default() -> Self {
Self {
capsule_id: None,
validate: true,
}
}
}
#[derive(Debug, Clone)]
pub struct CreatePinOptions {
pub target_type: PinTargetType,
pub target_id: Id,
pub note: Option<String>,
pub ttl_ms: Option<i64>,
pub scope_ids: Option<ScopeIds>,
}
pub struct KindlingService {
store: SqliteKindlingStore,
}
impl KindlingService {
pub fn new(store: SqliteKindlingStore) -> Self {
Self { store }
}
pub fn open(path: &Path) -> ServiceResult<Self> {
Ok(Self::new(SqliteKindlingStore::open(path)?))
}
pub fn open_with_options(path: &Path, options: &StoreOptions) -> ServiceResult<Self> {
Ok(Self::new(SqliteKindlingStore::open_with_options(
path, options,
)?))
}
pub fn open_in_memory() -> ServiceResult<Self> {
Ok(Self::new(SqliteKindlingStore::open_in_memory()?))
}
pub fn store(&self) -> &SqliteKindlingStore {
&self.store
}
pub fn open_capsule(&self, options: OpenCapsuleOptions) -> ServiceResult<Capsule> {
self.open_capsule_at(options, now_ms())
}
pub fn open_capsule_at(
&self,
options: OpenCapsuleOptions,
now: Timestamp,
) -> ServiceResult<Capsule> {
if options.kind == CapsuleType::Session {
if let Some(session_id) = options.scope_ids.session_id.as_deref() {
if let Some(existing) = self.store.get_open_capsule_for_session(session_id)? {
return Err(ServiceError::Conflict(format!(
"session {session_id} already has an open capsule ({})",
existing.id
)));
}
}
}
let capsule = validation::validate_capsule(
CapsuleInput {
id: options.id,
kind: options.kind,
intent: options.intent,
status: Some(CapsuleStatus::Open),
opened_at: None,
closed_at: None,
scope_ids: options.scope_ids,
observation_ids: None,
summary_id: None,
},
now,
)
.map_err(ServiceError::Validation)?;
self.store.create_capsule(&capsule)?;
Ok(capsule)
}
pub fn close_capsule(
&self,
capsule_id: &str,
options: CloseCapsuleOptions,
) -> ServiceResult<Capsule> {
self.close_capsule_at(capsule_id, options, now_ms())
}
pub fn close_capsule_at(
&self,
capsule_id: &str,
options: CloseCapsuleOptions,
now: Timestamp,
) -> ServiceResult<Capsule> {
let mut capsule = match self.store.get_capsule(capsule_id)? {
None => return Err(ServiceError::NotFound(capsule_id.to_string())),
Some(capsule) => capsule,
};
if capsule.status == CapsuleStatus::Closed {
return Err(ServiceError::AlreadyClosed(capsule_id.to_string()));
}
if options.generate_summary {
if let Some(content) = options.summary_content {
let summary = validation::validate_summary(
SummaryInput {
id: None,
capsule_id: capsule_id.to_string(),
content,
confidence: options.confidence.unwrap_or(1.0),
created_at: Some(now),
evidence_refs: Vec::new(),
},
now,
)
.map_err(ServiceError::Validation)?;
self.store.insert_summary(&summary)?;
}
}
match self.store.close_capsule(capsule_id, Some(now), None) {
Ok(()) => {}
Err(StoreError::CapsuleNotOpen(_)) => {
return Err(ServiceError::AlreadyClosed(capsule_id.to_string()))
}
Err(err) => return Err(err.into()),
}
capsule.status = CapsuleStatus::Closed;
capsule.closed_at = Some(now);
Ok(capsule)
}
pub fn append_observation(
&self,
input: ObservationInput,
options: AppendObservationOptions,
) -> ServiceResult<AppendOutcome> {
self.append_observation_at(input, options, now_ms())
}
pub fn append_observation_at(
&self,
input: ObservationInput,
options: AppendObservationOptions,
now: Timestamp,
) -> ServiceResult<AppendOutcome> {
let mut observation = if options.validate {
validation::validate_observation(input, now).map_err(ServiceError::Validation)?
} else {
validation::normalize_observation(input, now)
};
observation.content = mask_secrets(&observation.content);
let written = self.store.insert_observation(&observation)?;
let (observation, deduplicated) = if written {
(observation, false)
} else {
let existing = self
.store
.get_observation_by_id(&observation.id)?
.ok_or_else(|| {
ServiceError::Store(StoreError::ObservationNotFound(observation.id.clone()))
})?;
(existing, true)
};
if let Some(capsule_id) = options.capsule_id.as_deref() {
self.store
.attach_observation_to_capsule(capsule_id, &observation.id)?;
}
Ok(AppendOutcome {
observation,
deduplicated,
})
}
pub fn retrieve(&self, options: RetrieveOptions) -> ServiceResult<RetrieveResult> {
self.retrieve_at(options, now_ms())
}
pub fn retrieve_at(
&self,
options: RetrieveOptions,
now: Timestamp,
) -> ServiceResult<RetrieveResult> {
let provider = LocalFtsProvider::from_store(&self.store);
Ok(retrieve_at(&self.store, &provider, &options, now)?)
}
pub fn list_observations(
&self,
request: ListObservationsRequest,
) -> ServiceResult<ListObservationsResult> {
const DEFAULT_LIMIT: u32 = 100;
const MAX_LIMIT: u32 = 1000;
let limit = request.limit.unwrap_or(DEFAULT_LIMIT).clamp(1, MAX_LIMIT);
let cursor = match request.cursor.as_deref() {
Some(raw) => Some(cursor::decode(raw).map_err(|()| {
ServiceError::Validation(vec![ValidationError {
field: "cursor".to_string(),
message: "malformed pagination cursor".to_string(),
value: None,
}])
})?),
None => None,
};
let mut scope = request.scope_ids.clone();
scope.task_id = None;
let include_redacted = request.include_redacted.unwrap_or(false);
let cursor_ref = cursor.as_ref().map(|(ts, id)| (*ts, id.as_str()));
let fetch = limit.saturating_add(1);
let mut observations = self.store.list_observations(
Some(&scope),
&request.kinds,
request.since,
request.until,
cursor_ref,
include_redacted,
fetch,
)?;
let next_cursor = if observations.len() as u32 > limit {
observations.truncate(limit as usize);
observations.last().map(|o| cursor::encode(o.ts, &o.id))
} else {
None
};
Ok(ListObservationsResult {
observations,
next_cursor,
})
}
pub fn pin(&self, options: CreatePinOptions) -> ServiceResult<Pin> {
self.pin_at(options, now_ms())
}
pub fn pin_at(&self, options: CreatePinOptions, now: Timestamp) -> ServiceResult<Pin> {
let expires_at = options.ttl_ms.map(|ttl| now + ttl);
let pin = validation::validate_pin(
PinInput {
id: None,
target_type: options.target_type,
target_id: options.target_id,
reason: options.note,
created_at: Some(now),
expires_at,
scope_ids: options.scope_ids.unwrap_or_default(),
},
now,
)
.map_err(ServiceError::Validation)?;
self.store.insert_pin(&pin)?;
Ok(pin)
}
pub fn unpin(&self, pin_id: &str) -> ServiceResult<()> {
self.store.delete_pin(pin_id)?;
Ok(())
}
pub fn forget(&self, observation_id: &str) -> ServiceResult<()> {
self.store.redact_observation(observation_id)?;
Ok(())
}
pub fn get_capsule(&self, capsule_id: &str) -> ServiceResult<Option<Capsule>> {
Ok(self.store.get_capsule(capsule_id)?)
}
pub fn get_open_capsule(&self, session_id: &str) -> ServiceResult<Option<Capsule>> {
Ok(self.store.get_open_capsule_for_session(session_id)?)
}
pub fn get_observation(&self, observation_id: &str) -> ServiceResult<Option<Observation>> {
Ok(self.store.get_observation_by_id(observation_id)?)
}
pub fn get_summary(&self, summary_id: &str) -> ServiceResult<Option<Summary>> {
Ok(self.store.get_summary_by_id(summary_id)?)
}
pub fn get_latest_summary(&self, capsule_id: &str) -> ServiceResult<Option<Summary>> {
Ok(self.store.get_latest_summary_for_capsule(capsule_id)?)
}
pub fn list_pins(&self, scope: Option<&ScopeIds>) -> ServiceResult<Vec<Pin>> {
self.list_pins_at(scope, now_ms())
}
pub fn list_pins_at(
&self,
scope: Option<&ScopeIds>,
now: Timestamp,
) -> ServiceResult<Vec<Pin>> {
Ok(self.store.list_active_pins(scope, Some(now))?)
}
pub fn session_start_context(
&self,
scope: &ScopeIds,
max_results: u32,
) -> ServiceResult<SessionStartContext> {
self.session_start_context_at(scope, max_results, now_ms())
}
pub fn session_start_context_at(
&self,
scope: &ScopeIds,
max_results: u32,
now: Timestamp,
) -> ServiceResult<SessionStartContext> {
let pins = self.resolved_active_pins(scope, now)?;
let recent = self
.store
.query_observations(Some(scope), None, None, max_results)?;
Ok(SessionStartContext { pins, recent })
}
pub fn pre_compact_context(&self, scope: &ScopeIds) -> ServiceResult<PreCompactContext> {
self.pre_compact_context_at(scope, now_ms())
}
pub fn pre_compact_context_at(
&self,
scope: &ScopeIds,
now: Timestamp,
) -> ServiceResult<PreCompactContext> {
let pins = self.resolved_active_pins(scope, now)?;
let latest_summary = self
.store
.latest_summary_for_scope(Some(scope))?
.filter(|s| !s.content.is_empty());
Ok(PreCompactContext {
pins,
latest_summary,
})
}
fn resolved_active_pins(
&self,
scope: &ScopeIds,
now: Timestamp,
) -> ServiceResult<Vec<ResolvedPin>> {
let pins = self.store.list_active_pins(Some(scope), Some(now))?;
pins.into_iter()
.map(|pin| {
let content = match pin.target_type {
PinTargetType::Observation => self
.store
.get_observation_by_id(&pin.target_id)?
.map(|o| o.content),
PinTargetType::Summary => self
.store
.get_summary_by_id(&pin.target_id)?
.map(|s| s.content),
};
Ok(ResolvedPin {
note: pin.reason,
content,
})
})
.collect()
}
}
fn now_ms() -> Timestamp {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("system clock before Unix epoch")
.as_millis() as Timestamp
}
mod cursor {
use kindling_types::Timestamp;
const ALPHABET: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_";
fn b64_encode(input: &[u8]) -> String {
let mut out = String::with_capacity(input.len().div_ceil(3) * 4);
for chunk in input.chunks(3) {
let b0 = chunk[0] as u32;
let b1 = *chunk.get(1).unwrap_or(&0) as u32;
let b2 = *chunk.get(2).unwrap_or(&0) as u32;
let n = (b0 << 16) | (b1 << 8) | b2;
out.push(ALPHABET[((n >> 18) & 0x3f) as usize] as char);
out.push(ALPHABET[((n >> 12) & 0x3f) as usize] as char);
if chunk.len() > 1 {
out.push(ALPHABET[((n >> 6) & 0x3f) as usize] as char);
}
if chunk.len() > 2 {
out.push(ALPHABET[(n & 0x3f) as usize] as char);
}
}
out
}
fn b64_decode(input: &str) -> Result<Vec<u8>, ()> {
fn val(c: u8) -> Result<u32, ()> {
match c {
b'A'..=b'Z' => Ok((c - b'A') as u32),
b'a'..=b'z' => Ok((c - b'a' + 26) as u32),
b'0'..=b'9' => Ok((c - b'0' + 52) as u32),
b'-' => Ok(62),
b'_' => Ok(63),
_ => Err(()),
}
}
let bytes = input.as_bytes();
let mut out = Vec::with_capacity(input.len() / 4 * 3);
for chunk in bytes.chunks(4) {
if chunk.len() < 2 {
return Err(());
}
let mut n = 0u32;
for (i, &c) in chunk.iter().enumerate() {
n |= val(c)? << (18 - 6 * i);
}
out.push((n >> 16) as u8);
if chunk.len() > 2 {
out.push((n >> 8) as u8);
}
if chunk.len() > 3 {
out.push(n as u8);
}
}
Ok(out)
}
pub(crate) fn encode(ts: Timestamp, id: &str) -> String {
b64_encode(format!("{ts}:{id}").as_bytes())
}
pub(crate) fn decode(raw: &str) -> Result<(Timestamp, String), ()> {
let s = String::from_utf8(b64_decode(raw)?).map_err(|_| ())?;
let (ts_str, id) = s.split_once(':').ok_or(())?;
let ts = ts_str.parse::<Timestamp>().map_err(|_| ())?;
if id.is_empty() {
return Err(());
}
Ok((ts, id.to_string()))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn round_trips_various_lengths() {
for id in [
"a",
"ab",
"abc",
"abcd",
"550e8400-e29b-41d4-a716-446655440000",
] {
let token = encode(1_750_000_000_123, id);
assert_eq!(decode(&token).unwrap(), (1_750_000_000_123, id.to_string()));
}
}
#[test]
fn rejects_malformed() {
assert!(decode("").is_err()); assert!(decode("####").is_err()); assert!(decode(&b64_encode(b"no-colon-here")).is_err()); assert!(decode(&b64_encode(b"notanumber:abc")).is_err()); assert!(decode(&b64_encode(b"123:")).is_err()); }
}
}