asurada 0.1.0

Asurada — a memory + cognition daemon that grows with the user. Local-first, BYOK, shared by Devist/Webchemist Core/etc.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104

// brain.db ↔ Supabase Postgres 양방향 sync.

pub mod pull;
pub mod push;

use std::time::Duration;

use anyhow::{Context, Result};
use deadpool_postgres::{Config, Manager, ManagerConfig, Pool, RecyclingMethod};
use rusqlite::Connection;
use std::str::FromStr;
use std::sync::{Arc, Mutex};
use tokio_postgres::{Config as PgConfig, NoTls};

use crate::credentials::Credentials;

#[derive(Clone)]
pub struct Sync {
    pub brain: Arc<Mutex<Connection>>,
    pub pg: Pool,
    pub user_id: String,
}

impl Sync {
    pub async fn connect(brain: Arc<Mutex<Connection>>, creds: &Credentials) -> Result<Self> {
        // tokio-postgres Config 파싱.
        let pg_cfg = PgConfig::from_str(&creds.database_url)
            .context("parse ASURADA_DATABASE_URL")?;

        let mgr_cfg = ManagerConfig {
            recycling_method: RecyclingMethod::Fast,
        };
        let mgr = Manager::from_config(pg_cfg, NoTls, mgr_cfg);
        let pool = Pool::builder(mgr)
            .max_size(3)
            .build()
            .context("build pg pool")?;

        // 헬스체크: 스키마가 적용됐는지 확인.
        let client = pool.get().await.context("get pg client")?;
        let row = client
            .query_one(
                "SELECT EXISTS (
                   SELECT 1 FROM information_schema.tables
                   WHERE table_schema = 'asurada' AND table_name = 'events'
                 )",
                &[],
            )
            .await
            .context("check asurada schema")?;
        let exists: bool = row.get(0);
        if !exists {
            anyhow::bail!(
                "asurada 스키마가 Postgres 에 없습니다.\n\
                 다음을 한 번 실행해야 합니다:\n\
                 \tpsql \"$ASURADA_DATABASE_URL\" -f migrations/supabase/0001_init.sql"
            );
        }

        // unused 보다 명시: Config struct 가 deadpool 에서 활용되는데 import 필요.
        let _ = Config::new();

        Ok(Self {
            brain,
            pg: pool,
            user_id: creds.user_id.clone(),
        })
    }

    /// 백그라운드 sync 루프. `serve` 가 spawn 해서 데몬 라이프타임 동안 돌린다.
    /// 매 interval 마다 push 후 pull 한 번씩.
    /// 일시적 에러는 로그만 남기고 다음 라운드 진행 (네트워크 끊김 등).
    pub async fn run_loop(self, interval: Duration) {
        tracing::info!("sync loop started (interval={}s)", interval.as_secs());
        let mut ticker = tokio::time::interval(interval);
        ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
        // 첫 번째 tick 은 즉시 발생하므로 한 번 소비.
        ticker.tick().await;

        loop {
            ticker.tick().await;
            match self.push_all().await {
                Ok(p) if p.events + p.memories + p.advice + p.projects > 0 => {
                    tracing::info!(
                        "[sync] push events={} memories={} advice={} projects={}",
                        p.events, p.memories, p.advice, p.projects
                    );
                }
                Ok(_) => { /* nothing changed */ }
                Err(e) => tracing::warn!("[sync] push err: {}", e),
            }
            match self.pull_all().await {
                Ok(p) if p.events + p.memories + p.advice + p.projects > 0 => {
                    tracing::info!(
                        "[sync] pull events={} memories={} advice={} projects={}",
                        p.events, p.memories, p.advice, p.projects
                    );
                }
                Ok(_) => {}
                Err(e) => tracing::warn!("[sync] pull err: {}", e),
            }
        }
    }
}