ryan/vault-cli
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
d89aeb7d261b6cd0e09c6db2a0a6a66573a2f921
vault-cli/vault_sync_daemon.py
T

515 lines
20 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""
vault_sync_daemon.py — long-running inotify (watchdog) daemon that keeps
/vault (JDex notes, git-backed) and /data (blob storage, Duplicati-backed)
structurally mirrored, in BOTH directions, for changes made by ANY tool
(Obsidian, SMB client, manual mv, not just the `vault` CLI).
Rules (confirmed with Ryan 2026-09-15):
FORWARD (/vault -> /data):
- rename/move of an ID folder in /vault -> mirror the rename in /data.
- delete of an ID folder in /vault -> do NOT delete /data's copy.
Instead: move the /data folder into the NEAREST existing `.09` archive
folder up the tree (category-level first, then area/system; create a
category-level `.09 Archive` in /data if none exists at any level yet),
and append a note to the nearest `.09` archive's /vault folder-note
markdown recording what was archived, when, and where.
REVERSE (/data -> /vault):
- a new folder appears in /data whose basename matches the `SYS.AC.ID `
Johnny Decimal pattern, with NO /vault counterpart yet (e.g. dropped in
over SMB) -> auto-create a minimal /vault stub folder + folder-note,
tagged "needs review", AND append a `#fix-me` line to the nearest `.01`
Inbox note in /vault (category-level first, then area/system, falling
back to the top-level System inbox).
- Folders inside any `.09`-pattern archive folder are never auto-stubbed
(archives are meant to hold retired stuff, not grow new tracked IDs) —
this also breaks the feedback loop from the forward-delete archive move.
- Generic non-ID subfolders (e.g. "Documents/" inside an existing ID
folder) are ignored entirely — normal blob organization, not JD structure.
Idempotent by construction: every action checks target state first, so a
duplicate/replayed event from the daemon's own writes is a safe no-op.
Run: python3 vault_sync_daemon.py (foreground; systemd unit wraps this)
Logs: /home/hermes/.hermes/logs/vault-sync-daemon.log
"""
import os
import re
import sys
import time
import shutil
import subprocess
import datetime
import threading
_SELF_DIR = os.path.dirname(os.path.realpath(__file__))
sys.path.insert(0, _SELF_DIR)
import vault_lib as vl
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler, DirCreatedEvent, DirMovedEvent, DirDeletedEvent
LOG_PATH = "/home/hermes/.hermes/logs/vault-sync-daemon.log"
os.makedirs(os.path.dirname(LOG_PATH), exist_ok=True)
NTFY_TOKEN_PATH = "/home/hermes/.hermes/secrets/ntfy-vault-token"
NTFY_URL = "https://ntfy.pandu.ski/vault"
def ntfy_post(title, message, tags="package", priority="default"):
"""Fire-and-forget ntfy notification. Never raises — a notify failure
must never block or crash the daemon's actual sync work."""
try:
token = open(NTFY_TOKEN_PATH, encoding="utf-8").read().strip()
except Exception as e:
log(f"ntfy_post: could not read token: {e}")
return
try:
subprocess.run(
["curl", "-fsS", "--max-time", "5", "-X", "POST",
"-H", f"Authorization: Bearer {token}",
"-H", f"Title: {title}",
"-H", f"Tags: {tags}",
"-H", f"Priority: {priority}",
"-d", message,
NTFY_URL],
check=False, capture_output=True, timeout=8,
)
except Exception as e:
log(f"ntfy_post failed: {e}")
VAULT = vl.VAULT
DATA = vl.DATA
SKIP_DIRS = vl.SKIP_DIRS
DATA_EXEMPT = vl.DATA_EXEMPT
ID_RE = vl.ID_RE
CAT_RE = vl.CAT_RE
ARCHIVE_ID_RE = re.compile(r'^([RJFI])\.(\d\d)\.09 ') # e.g. "F.25.09 Archive"
ARCHIVE_CAT_RE = re.compile(r'^([RJFI])\.(\d\d) ') # category folder, to find/create .09 inside
INBOX_ID_RE = re.compile(r'^([RJFI])\.(\d\d)\.01 ')
_lock = threading.Lock()
_recent_events = {} # (kind, path) -> timestamp, for debouncing duplicate inotify events
_DEBOUNCE_SECONDS = 3.0
def _debounced(kind, path):
"""
Return True (and record) if this (kind, path) event was NOT seen in the
last _DEBOUNCE_SECONDS — inotify/watchdog can emit duplicate CREATE
events for a single mkdir, so every handler must be idempotent AND
debounced to avoid double-processing (e.g. flagging the same new /data
folder in two different inbox notes).
"""
now = time.time()
key = (kind, path)
last = _recent_events.get(key)
_recent_events[key] = now
# opportunistic cleanup so this dict never grows unbounded
if len(_recent_events) > 2000:
cutoff = now - _DEBOUNCE_SECONDS
for k, v in list(_recent_events.items()):
if v < cutoff:
del _recent_events[k]
return last is None or (now - last) > _DEBOUNCE_SECONDS
def log(msg):
line = f"{datetime.datetime.now().isoformat(timespec='seconds')} {msg}"
print(line, flush=True)
try:
with open(LOG_PATH, "a", encoding="utf-8") as f:
f.write(line + "\n")
except Exception:
pass
def _git_commit(msg, paths=None):
"""
Commit only the specific paths the daemon itself touched — never a
blanket `git add -A`, so a human's concurrent in-progress edits
elsewhere in the vault are never swept into an unrelated auto-commit.
"""
if not paths:
return
subprocess.run(["git", "-C", VAULT, "add", "--"] + list(paths), check=False)
diff = subprocess.run(["git", "-C", VAULT, "diff", "--cached", "--quiet"], check=False)
if diff.returncode == 0:
return # nothing staged
subprocess.run(["git", "-C", VAULT, "commit", "-m", msg], check=False)
def _in_skip(path, root):
rel = os.path.relpath(path, root)
parts = rel.split(os.sep)
return any(p in SKIP_DIRS for p in parts)
def _under_archive(rel_path):
"""True if any path component matches the '<SYS.AC>.09 ' archive pattern."""
for part in rel_path.split(os.sep):
if ARCHIVE_ID_RE.match(part):
return True
return False
def _under_data_exempt(rel_path):
top = rel_path.split(os.sep)[0]
return top in DATA_EXEMPT
# ---------------------------------------------------------------------------
# FORWARD: /vault -> /data
# ---------------------------------------------------------------------------
def handle_vault_created(path):
"""
A new AREA/CATEGORY/ID-shaped folder appeared in /vault (e.g. created
via Obsidian, not the `vault` CLI) with no /data counterpart yet.
Mirror it: create the matching empty folder in /data. This is the
forward-create leg that was missing — only move/delete were wired up
originally, so a brand-new folder (like "R.6 Test") never propagated.
"""
if not os.path.isdir(path):
return
if _in_skip(path, VAULT):
return
if not _debounced("vault_created", path):
return
rel = os.path.relpath(path, VAULT)
base = os.path.basename(path)
if not (ID_RE.match(base) or CAT_RE.match(base) or vl.AREA_RE.match(base)):
return # not a JD-shaped folder — ignore (e.g. "Documents" subfolder)
data_path = os.path.join(DATA, rel)
if os.path.isdir(data_path):
return # already mirrored (duplicate event or pre-existing)
parent = os.path.dirname(data_path)
if not os.path.isdir(parent):
log(f"SKIP create mirror (parent missing in /data): {rel}")
return
os.makedirs(data_path, exist_ok=True)
log(f"MIRRORED create /data: {rel}")
ntfy_post(
"Vault sync: new folder",
f"Created in /vault, mirrored to /data: {rel}",
tags="sparkles",
)
def handle_vault_moved(src_path, dest_path):
if _in_skip(src_path, VAULT) or _in_skip(dest_path, VAULT):
return
rel_src = os.path.relpath(src_path, VAULT)
rel_dest = os.path.relpath(dest_path, VAULT)
data_src = os.path.join(DATA, rel_src)
data_dest = os.path.join(DATA, rel_dest)
if not os.path.isdir(data_src):
return # nothing to mirror
if os.path.exists(data_dest):
log(f"SKIP move mirror (dest exists): {rel_src} -> {rel_dest}")
return
os.makedirs(os.path.dirname(data_dest), exist_ok=True)
os.rename(data_src, data_dest)
log(f"MIRRORED rename /data: {rel_src} -> {rel_dest}")
ntfy_post(
"Vault sync: renamed",
f"/data: {rel_src} -> {rel_dest}",
tags="package",
)
def _category_title(catname):
"""'F.22 Fun lists' -> 'Fun lists'"""
m = CAT_RE.match(catname)
return catname[m.end():].strip() if m else catname
def _find_nearest_archive_dir(rel_path):
"""
Archive destination is ALWAYS the immediate owning category of the
deleted ID (one level up from the ID folder) — never a higher-level
ancestor's archive, even if one already exists there. If that category
has no '<SYS.AC>.09 ' archive folder yet, create it right there
(materializing the missing JD archive slot), matching Ryan's rule:
"create missing .01s and .09s when they're missing", not borrow one
further up the tree.
Returns (archive_data_path, vault_note_path_or_None).
vault_note_path is set only if this call just created a NEW archive
slot's folder-note (so the caller knows to git-add it).
"""
parts = rel_path.split(os.sep)
if len(parts) < 2:
return (DATA, None) # ID sits at top level (shouldn't happen)
cat_rel = os.sep.join(parts[:-1])
cat_vault = os.path.join(VAULT, cat_rel)
cat_data = os.path.join(DATA, cat_rel)
catname = os.path.basename(cat_vault)
m = CAT_RE.match(catname)
if not m or not os.path.isdir(cat_vault):
return (cat_data if os.path.isdir(cat_data) else DATA, None)
sysletter, catnum = m.group(1), m.group(2)
# already has an archive slot in this category?
if os.path.isdir(cat_vault):
for name in sorted(os.listdir(cat_vault)):
if ARCHIVE_ID_RE.match(name):
dpath = os.path.join(cat_data, name)
os.makedirs(dpath, exist_ok=True)
return (dpath, None)
# missing — create it in THIS category, not a higher one
archive_name = f"{sysletter}.{catnum}.09 Archive for {_category_title(catname)}"
vpath = os.path.join(cat_vault, archive_name)
vnote = os.path.join(vpath, archive_name + ".md")
created_note = None
if not os.path.exists(vnote):
os.makedirs(vpath, exist_ok=True)
with open(vnote, "w", encoding="utf-8") as f:
f.write(f"# {archive_name}\n\n"
f"Auto-created by vault_sync_daemon on "
f"{datetime.datetime.now().isoformat(timespec='seconds')} "
f"as the archive destination for a deleted ID in this category.\n")
log(f"CREATED category archive slot: {os.path.relpath(vpath, VAULT)}")
created_note = vnote
archive_path = os.path.join(cat_data, archive_name)
os.makedirs(archive_path, exist_ok=True)
return (archive_path, created_note)
def _nearest_archive_note(archive_data_path):
"""Given an archive dir in /data, return the matching /vault folder-note path, if any."""
rel = os.path.relpath(archive_data_path, DATA)
vault_dir = os.path.join(VAULT, rel)
if not os.path.isdir(vault_dir):
return None
note = os.path.join(vault_dir, os.path.basename(vault_dir) + ".md")
return note if os.path.exists(note) else None
def handle_vault_deleted(path):
if _in_skip(path, VAULT):
return
if not _debounced("vault_deleted", path):
return
rel = os.path.relpath(path, VAULT)
base = os.path.basename(path)
if not ID_RE.match(base):
return # only care about ID-folder deletions
data_path = os.path.join(DATA, rel)
if not os.path.isdir(data_path):
return # already handled (duplicate event) or no /data counterpart ever existed
archive_dir, created_note = _find_nearest_archive_dir(rel)
dest = os.path.join(archive_dir, base)
if os.path.exists(dest):
dest = os.path.join(archive_dir, f"{base} (archived {int(time.time())})")
shutil.move(data_path, dest)
log(f"ARCHIVED /data: {rel} -> {os.path.relpath(dest, DATA)}")
note = _nearest_archive_note(archive_dir)
ts = datetime.datetime.now().isoformat(timespec="seconds")
entry = (f"\n- **Archived {ts}:** `{base}` (deleted from /vault) — "
f"data preserved at `/data/{os.path.relpath(dest, DATA)}`\n")
commit_paths = []
if note:
with open(note, "a", encoding="utf-8") as f:
f.write(entry)
log(f"NOTED archive in: {os.path.relpath(note, VAULT)}")
commit_paths.append(note)
if created_note:
commit_paths.append(created_note)
_git_commit(f"vault_sync_daemon: archived {base}", commit_paths)
ntfy_post(
"Vault sync: archived",
f"Deleted from /vault: {base}\nData preserved at /data/{os.path.relpath(dest, DATA)}",
tags="package,wastebasket",
)
# ---------------------------------------------------------------------------
# REVERSE: /data -> /vault
# ---------------------------------------------------------------------------
def _find_nearest_inbox_note(rel_path):
"""
Inbox destination is ALWAYS the immediate owning category of the new
/data folder (same rule as archives: create the missing '.01 Inbox'
right there if it doesn't exist yet, never borrow one from a higher
ancestor).
"""
parts = rel_path.split(os.sep)
if len(parts) < 2:
return None
cat_rel = os.sep.join(parts[:-1])
cat_vault = os.path.join(VAULT, cat_rel)
cat_data = os.path.join(DATA, cat_rel)
catname = os.path.basename(cat_vault)
m = CAT_RE.match(catname)
if not m or not os.path.isdir(cat_vault):
return None
sysletter, catnum = m.group(1), m.group(2)
for name in sorted(os.listdir(cat_vault)):
if INBOX_ID_RE.match(name):
note = os.path.join(cat_vault, name, name + ".md")
if os.path.exists(note):
return note
# missing — create it in THIS category, not a higher one
inbox_name = f"{sysletter}.{catnum}.01 Inbox for {_category_title(catname)}"
vpath = os.path.join(cat_vault, inbox_name)
note = os.path.join(vpath, inbox_name + ".md")
os.makedirs(vpath, exist_ok=True)
with open(note, "w", encoding="utf-8") as f:
f.write(f"# {inbox_name}\n\n"
f"Auto-created by vault_sync_daemon on "
f"{datetime.datetime.now().isoformat(timespec='seconds')} "
f"to flag an unrecognized /data folder in this category.\n")
log(f"CREATED category inbox slot: {os.path.relpath(vpath, VAULT)}")
dpath = os.path.join(cat_data, inbox_name)
os.makedirs(dpath, exist_ok=True)
return note
def handle_data_created(path):
if not os.path.isdir(path):
return
if _in_skip(path, DATA):
return
if not _debounced("data_created", path):
return # duplicate inotify event for the same mkdir, ignore
rel = os.path.relpath(path, DATA)
if _under_data_exempt(rel):
return
if _under_archive(rel):
return # never auto-stub inside archive folders
base = os.path.basename(path)
if not ID_RE.match(base):
return # only ID-shaped folders trigger reverse-sync
vault_path = os.path.join(VAULT, rel)
if os.path.isdir(vault_path):
return # already has a /vault counterpart
catcode_root = os.path.dirname(vault_path)
if not os.path.isdir(catcode_root):
log(f"SKIP stub (parent category folder missing in /vault): {rel}")
return
note = os.path.join(vault_path, base + ".md")
os.makedirs(vault_path, exist_ok=True)
ts = datetime.datetime.now().isoformat(timespec="seconds")
with open(note, "w", encoding="utf-8") as f:
f.write(f"# {base}\n\n"
f"> [!warning] Needs review\n"
f"> Auto-created by vault_sync_daemon on {ts} because a matching "
f"folder appeared in `/data` with no `/vault` note yet.\n")
log(f"STUBBED /vault note for new /data folder: {rel}")
commit_paths = [note]
inbox = _find_nearest_inbox_note(rel)
if inbox:
with open(inbox, "a", encoding="utf-8") as f:
f.write(f"\n- #fix-me New `/data` folder detected with no vault note: "
f"[[{base}]] (auto-stub created {ts})\n")
log(f"FLAGGED in inbox: {os.path.relpath(inbox, VAULT)}")
commit_paths.append(inbox)
_git_commit(f"vault_sync_daemon: stubbed {base}", commit_paths)
ntfy_post(
"Vault sync: needs review",
f"New /data folder with no vault note: {rel}\nStub created, flagged in nearest inbox.",
tags="package,warning",
priority="high",
)
# ---------------------------------------------------------------------------
# watchdog wiring
# ---------------------------------------------------------------------------
class VaultHandler(FileSystemEventHandler):
def on_created(self, event):
if not event.is_directory:
return
with _lock:
try:
time.sleep(0.5) # let Obsidian finish writing the folder-note
handle_vault_created(event.src_path)
except Exception as e:
log(f"ERROR handle_vault_created: {e}")
ntfy_post(
"Vault sync ERROR",
f"handle_vault_created failed:\n{event.src_path}\n{e}",
tags="rotating_light",
priority="urgent",
)
def on_moved(self, event):
if not event.is_directory:
return
with _lock:
try:
handle_vault_moved(event.src_path, event.dest_path)
except Exception as e:
log(f"ERROR handle_vault_moved: {e}")
ntfy_post(
"Vault sync ERROR",
f"handle_vault_moved failed:\n{event.src_path} -> {event.dest_path}\n{e}",
tags="rotating_light",
priority="urgent",
)
def on_deleted(self, event):
if not event.is_directory:
return
with _lock:
try:
handle_vault_deleted(event.src_path)
except Exception as e:
log(f"ERROR handle_vault_deleted: {e}")
ntfy_post(
"Vault sync ERROR",
f"handle_vault_deleted failed:\n{event.src_path}\n{e}",
tags="rotating_light",
priority="urgent",
)
class DataHandler(FileSystemEventHandler):
def on_created(self, event):
if not event.is_directory:
return
with _lock:
try:
# small settle delay so a batch mkdir/copy isn't caught half-written
time.sleep(0.5)
handle_data_created(event.src_path)
except Exception as e:
log(f"ERROR handle_data_created: {e}")
ntfy_post(
"Vault sync ERROR",
f"handle_data_created failed:\n{event.src_path}\n{e}",
tags="rotating_light",
priority="urgent",
)
def main():
log("vault_sync_daemon starting")
observer = Observer()
observer.schedule(VaultHandler(), VAULT, recursive=True)
observer.schedule(DataHandler(), DATA, recursive=True)
observer.start()
log(f"watching {VAULT} and {DATA}")
ntfy_post("Vault sync daemon started", f"Watching {VAULT} and {DATA}", tags="rocket", priority="low")
try:
while True:
time.sleep(5)
except KeyboardInterrupt:
pass
finally:
observer.stop()
observer.join()
log("vault_sync_daemon stopped")
# NOTE: no ntfy_post here — a hard kill/OOM/crash skips this `finally`
# entirely, so relying on it for "stopped" alerts is unreliable. The
# systemd unit's ExecStopPost= hook fires unconditionally on every
# exit (clean or not) and is the actual source of stop notifications.
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink