Compare commits

..
7 Commits
Author SHA1 Message Date
Worph 0403a6f3bc replace file discovery by udp discovery
Build and Push Docker Image / build (push) Failing after 13s
2026-09-14 11:24:46 +02:00
Worph 355c666183 chnage service discovery and identity 2026-09-13 19:48:52 +02:00
Worph 0cccceccd9 align with metadata standard
Build and Push Docker Image / build (push) Failing after 13s
2026-09-13 15:57:30 +02:00
Worph 40b553016c cid vector
Build and Push Docker Image / build (push) Failing after 13s
2026-09-13 14:36:50 +02:00
Worph 4894810fd4 new file tuple API
Build and Push Docker Image / build (push) Failing after 13s
2026-09-13 12:01:33 +02:00
Worph ac21315255 improved cid resolution
Build and Push Docker Image / build (push) Failing after 13s
2026-09-12 21:37:38 +02:00
Worph ec1b227631 improved search 2026-09-12 16:37:51 +02:00
41 changed files with 3097 additions and 999 deletions
+5
View File
@@ -209,6 +209,7 @@ the `/file/{cid}` reverse-lookup path.)
```bash
GET /api/metadata/hash-ids
GET /api/files/tuples
GET /api/metadata/list
POST /api/metadata/search
POST /api/metadata/batch
@@ -287,6 +288,10 @@ return a deprecation pointer at the `/api/watchers/*` routes
```bash
POST /api/admin/migrate-dual-roots # reunify stranded midhash-rooted entries
POST /api/admin/sweep-literature-echoes[?apply=true]
# strip query echoes off literature cards: anilistid on site cards,
# workcid/chapterNumber/volumeNumber/chapterStart/chapterEnd on any card.
# Dry-run unless apply=true; returns {dryRun, planned, fixed, byKey}
GET/PUT/DELETE /webdav/... # mounted on /files
```
+55 -16
View File
@@ -10,8 +10,8 @@ import (
"github.com/metazla/meta-core/internal/api"
"github.com/metazla/meta-core/internal/config"
"github.com/metazla/meta-core/internal/discovery"
"github.com/metazla/meta-core/internal/leader"
"github.com/metazla/meta-core/internal/meshdisco"
"github.com/metazla/meta-core/internal/storage"
)
@@ -60,20 +60,47 @@ func main() {
}
log.Printf("[meta-core] Redis schema version OK (v%d)", storage.SchemaVersion)
// Create and start service discovery
disc := discovery.NewService(cfg)
if err := disc.Start(); err != nil {
log.Fatalf("[meta-core] Failed to start service discovery: %v", err)
// UDP service discovery (meta-discovery v1). Runs alongside the file
// registry above for now — nothing consumes the wire yet, so this cannot
// regress the existing path. The announce payload is rebuilt from
// leaderProvider on every tick, which is what keeps it and GET /urls from
// drifting apart.
var mesh *meshdisco.Node
if cfg.EnableUDPDiscovery {
hostname, _ := os.Hostname()
mesh = meshdisco.New(meshdisco.Config{
Name: cfg.ServiceName,
Instance: hostname,
Role: meshdisco.RoleCore,
Version: Version,
Group: cfg.DiscoveryGroup,
Port: cfg.DiscoveryPort,
Interval: time.Duration(cfg.DiscoveryIntervalMS) * time.Millisecond,
Payload: func() (string, string, *meshdisco.URLs) {
info := leaderProvider.LeaderInfo()
if info == nil {
return "", "starting", nil
}
// Create and start dead service cleaner
cleaner := discovery.NewCleaner(cfg)
if err := cleaner.Start(); err != nil {
log.Printf("[meta-core] Warning: failed to start service cleaner: %v", err)
return coreBaseURL(cfg), "running", &meshdisco.URLs{
Hostname: info.Hostname,
BaseUrl: info.BaseUrl,
ApiUrl: info.ApiUrl,
WebdavUrl: info.WebdavUrl,
WebdavUrlInternal: info.WebdavUrlInternal,
}
},
})
if err := mesh.Start(); err != nil {
log.Printf("[meta-core] Warning: UDP discovery disabled: %v", err)
mesh = nil
}
} else {
log.Println("[meta-core] UDP discovery disabled by ENABLE_UDP_DISCOVERY")
}
// Create and start API server
apiServer := api.NewServer(cfg, leaderProvider, disc, cleaner, storageClient)
apiServer := api.NewServer(cfg, leaderProvider, storageClient)
apiServer.SetMeshNode(mesh)
if err := apiServer.Start(); err != nil {
log.Fatalf("[meta-core] Failed to start API server: %v", err)
}
@@ -90,12 +117,10 @@ func main() {
log.Printf("[meta-core] Error stopping API server: %v", err)
}
if err := cleaner.Stop(); err != nil {
log.Printf("[meta-core] Error stopping service cleaner: %v", err)
if mesh != nil {
if err := mesh.Stop(); err != nil {
log.Printf("[meta-core] Error stopping UDP discovery: %v", err)
}
if err := disc.Stop(); err != nil {
log.Printf("[meta-core] Error stopping service discovery: %v", err)
}
if err := storageClient.Close(); err != nil {
@@ -112,3 +137,17 @@ func waitForShutdown() {
<-sigChan
}
// coreBaseURL is the browser-facing URL announced for the nav menu. It keeps
// the three-way fallback the file registry used (discovery/service.go):
// META_CORE_PUBLIC_URL wins (the reachable URL when there is no Caddy
// perimeter, e.g. a debug-direct port), then BASE_URL (the Caddy URL), then
// the container IP.
func coreBaseURL(cfg *config.Config) string {
if cfg.PublicURL != "" {
return cfg.PublicURL
}
if cfg.BaseURL != "" {
return cfg.BaseURL
}
return fmt.Sprintf("http://%s:%d", meshdisco.LocalIPv4(), cfg.APIPort)
}
+38 -140
View File
@@ -1,155 +1,53 @@
import { useState, useEffect } from 'react';
import { useEffect } from 'react';
// eslint-disable-next-line @typescript-eslint/ban-ts-comment
// @ts-ignore -- plain ES module, no .d.ts; imported for its side effect
// of defining the custom element.
import '../meta-service-menu.js';
interface ServiceInfo {
name: string;
hostname: string;
baseUrl: string;
status: string;
lastHeartbeat: string;
}
/**
* ServiceNav — the neighbouring meta-* services, as a burger menu.
*
* Since meta-discovery v1 this is a thin wrapper around <meta-service-menu>,
* a framework-free custom element shared by every meta-* UI (see
* scripts/check-mirrors.sh). The old implementation was ~165 lines of React
* that existed in four near-identical copies and polled meta-core's
* /api/services; the element polls this service's own /api/neighbors instead,
* so the nav still renders when meta-core is down.
*
* The element lives in shadow DOM and cannot see this app's CSS variables, so
* the four --mm-nav-* tokens below map our palette onto it.
*/
interface ServicesResponse {
services: ServiceInfo[];
current: string;
}
const serviceIcons: Record<string, string> = {
'meta-core': '🗄️',
'meta-sort': '📁',
'meta-fuse': '🗂️',
'meta-stremio': '🎬',
'meta-orbit': '🌐',
'default': '📦'
};
function formatServiceName(name: string): string {
return name.split('-').map(word =>
word.charAt(0).toUpperCase() + word.slice(1)
).join(' ');
declare global {
// eslint-disable-next-line @typescript-eslint/no-namespace
namespace JSX {
interface IntrinsicElements {
'meta-service-menu': React.DetailedHTMLProps<
React.HTMLAttributes<HTMLElement>,
HTMLElement
> & { current?: string; endpoint?: string; label?: string };
}
}
}
function ServiceNav() {
const [services, setServices] = useState<ServiceInfo[]>([]);
const [loading, setLoading] = useState(true);
const currentService = 'meta-core';
useEffect(() => {
const fetchServices = async () => {
try {
const response = await fetch('/services');
if (response.ok) {
const data: ServicesResponse = await response.json();
setServices(data.services || []);
}
} catch (error) {
console.error('Failed to fetch services:', error);
} finally {
setLoading(false);
}
};
fetchServices();
const interval = setInterval(fetchServices, 30000);
return () => clearInterval(interval);
// Import above is for the side effect of defining the element; this
// keeps bundlers from tree-shaking it away in production builds.
}, []);
if (loading) return null;
const sortedServices = [...services].sort((a, b) => a.name.localeCompare(b.name));
if (sortedServices.length === 0) return null;
return (
<nav className="services-nav">
<span className="services-nav-label">Services:</span>
<div className="services-nav-items">
{sortedServices.map(service => {
const icon = serviceIcons[service.name] || serviceIcons.default;
const isActive = service.name === currentService;
return (
<a
key={service.name}
href={isActive ? '#' : service.baseUrl}
className={`service-link${isActive ? ' active' : ''}`}
onClick={(e) => {
e.preventDefault();
if (!isActive) {
window.location.href = service.baseUrl;
}
}}
>
<span className="service-icon">{icon}</span>
<span>{formatServiceName(service.name)}</span>
<span className="service-status"></span>
</a>
);
})}
</div>
<>
<meta-service-menu current="meta-core" />
<style>{`
.services-nav {
display: flex;
align-items: center;
gap: 0.5rem;
}
.services-nav-label {
color: #888;
font-size: 0.8rem;
white-space: nowrap;
}
.services-nav-items {
display: flex;
gap: 0.35rem;
}
.service-link {
display: inline-flex;
align-items: center;
gap: 0.3rem;
padding: 0.35rem 0.6rem;
border-radius: 6px;
text-decoration: none;
font-size: 0.8rem;
font-weight: 500;
transition: all 0.2s;
color: #ccc;
background: #0f3460;
border: 1px solid #1a4a7a;
}
.service-link:hover {
background: #1a4a7a;
border-color: rgba(78, 205, 196, 0.5);
text-decoration: none;
color: #fff;
}
.service-link.active {
background: linear-gradient(135deg, rgba(78, 205, 196, 0.2), rgba(68, 160, 141, 0.2));
border-color: rgba(78, 205, 196, 0.5);
color: #fff;
}
.service-link .service-icon {
font-size: 0.9rem;
}
.service-link .service-status {
width: 5px;
height: 5px;
border-radius: 50%;
background: #4ecdc4;
}
@media (max-width: 800px) {
.services-nav-label {
display: none;
}
meta-service-menu {
--mm-nav-fg: var(--text-primary, #e0e0e0);
--mm-nav-bg: var(--bg-tertiary, var(--bg-secondary, #1a1a2e));
--mm-nav-border: var(--border-color, rgba(255,255,255,0.14));
--mm-nav-accent: var(--accent-primary, #4ecdc4);
}
`}</style>
</nav>
</>
);
}
+303
View File
@@ -0,0 +1,303 @@
/**
* <meta-service-menu> — the neighbouring meta-* services, as a burger menu.
*
* Replaces the eight divergent "services bar" implementations (four copies of
* a React `ServiceNav.tsx`, two mirrored `services.js`, two bespoke inline
* ones) with a single framework-free custom element. It drops into the React
* dashboards as a plain tag — no React dependency — and into the vanilla UIs
* as an ES module.
*
* Data comes from the host service's own `GET /api/neighbors`, same-origin.
* Never cross-origin: the dev stack's self-signed certs and the OIDC redirects
* make a fetch to a neighbour a dead end. Links are plain navigations.
*
* ⚠ MIRRORED FILE. Byte-identical copies live in every meta-* UI tree; see
* `scripts/check-mirrors.sh`. Edit one → edit all.
*
* Shadow DOM is not optional here. The eight UIs use four incompatible token
* vocabularies, and two of them define *different colours under the same
* names* (`--bg-secondary` is #12121a in meta-sort, #16213e in meta-core,
* #161b22 in meta-dup). meta-search and meta-share are also the only
* light-mode-capable UIs, so a hardcoded dark palette would break them.
* The element therefore styles itself from four tokens with dark defaults:
*
* --mm-nav-fg, --mm-nav-bg, --mm-nav-border, --mm-nav-accent
*
* Each host maps its own variables onto those, e.g.
* meta-service-menu { --mm-nav-bg: var(--bg-secondary); }
*
* Attributes:
* current name of the host service, so it can be marked and not linked
* endpoint override the default "/api/neighbors"
* label button tooltip (default "Services")
*
* Property:
* rewriteUrl optional (name, url, neighbour) => url, for hosts that must
* translate internal hostnames into browser-reachable ones
* (meta-gateway does this for its dev/prod split).
*/
const ICONS = {
"meta-core": "⚙️",
"meta-sort": "📁",
"meta-fuse": "🗂️",
"meta-stremio": "🎬",
"meta-dup": "🔍",
"meta-search": "🔎",
"meta-share": "🔗",
"meta-gateway": "🌉",
"meta-watch": "📺",
"meta-listen": "🎧",
"meta-read": "📖",
};
const DEFAULT_ICON = "📦";
const POLL_MS = 30000;
function titleCase(name) {
return String(name || "")
.split("-")
.map((w) => w.charAt(0).toUpperCase() + w.slice(1))
.join(" ");
}
const TEMPLATE = `
<style>
:host {
--_fg: var(--mm-nav-fg, #e8eaed);
--_bg: var(--mm-nav-bg, #1a1a2e);
--_border: var(--mm-nav-border, rgba(255, 255, 255, 0.14));
--_accent: var(--mm-nav-accent, #4ecdc4);
position: relative;
display: inline-block;
font-family: inherit;
font-size: 14px;
line-height: 1.4;
}
button.trigger {
display: inline-flex;
align-items: center;
justify-content: center;
gap: 0.4rem;
width: 2.1rem;
height: 2.1rem;
padding: 0;
cursor: pointer;
color: var(--_fg);
background: transparent;
border: 1px solid var(--_border);
border-radius: 8px;
font: inherit;
transition: border-color 0.15s, background 0.15s;
}
button.trigger:hover { border-color: var(--_accent); }
button.trigger:focus-visible {
outline: 2px solid var(--_accent);
outline-offset: 2px;
}
button.trigger[aria-expanded="true"] { border-color: var(--_accent); }
.bars { display: block; width: 15px; }
.bars span {
display: block;
height: 2px;
margin: 3px 0;
border-radius: 2px;
background: currentColor;
}
.panel {
position: absolute;
top: calc(100% + 6px);
right: 0;
z-index: 1000;
min-width: 13rem;
max-height: 70vh;
overflow-y: auto;
padding: 0.3rem;
background: var(--_bg);
border: 1px solid var(--_border);
border-radius: 10px;
box-shadow: 0 10px 30px rgba(0, 0, 0, 0.35);
}
.panel[hidden] { display: none; }
a.item, .item {
display: flex;
align-items: center;
gap: 0.55rem;
padding: 0.45rem 0.6rem;
border-radius: 6px;
color: var(--_fg);
text-decoration: none;
white-space: nowrap;
}
a.item:hover { background: color-mix(in srgb, var(--_accent) 18%, transparent); }
.item.active { color: var(--_accent); font-weight: 600; cursor: default; }
.item.dead { opacity: 0.55; cursor: not-allowed; }
.icon { font-size: 1rem; line-height: 1; }
.dot {
width: 6px;
height: 6px;
margin-left: auto;
border-radius: 50%;
background: var(--_accent);
}
.empty {
padding: 0.6rem;
opacity: 0.7;
font-size: 0.85em;
white-space: nowrap;
}
/* Respect a host that opts out of motion. */
@media (prefers-reduced-motion: reduce) {
button.trigger { transition: none; }
}
</style>
<button class="trigger" type="button" aria-haspopup="menu" aria-expanded="false">
<span class="bars" aria-hidden="true"><span></span><span></span><span></span></span>
</button>
<div class="panel" role="menu" hidden></div>
`;
class MetaServiceMenu extends HTMLElement {
constructor() {
super();
this.attachShadow({ mode: "open" });
this.shadowRoot.innerHTML = TEMPLATE;
this._btn = this.shadowRoot.querySelector("button.trigger");
this._panel = this.shadowRoot.querySelector(".panel");
this._neighbors = [];
this._timer = null;
this._open = false;
this._onDocClick = (e) => {
if (!this._open) return;
if (e.composedPath().includes(this)) return;
this._setOpen(false);
};
this._onKey = (e) => {
if (e.key === "Escape" && this._open) {
this._setOpen(false);
this._btn.focus();
}
};
}
connectedCallback() {
this._btn.title = this.getAttribute("label") || "Services";
this._btn.setAttribute("aria-label", this._btn.title);
this._btn.addEventListener("click", () => this._setOpen(!this._open));
document.addEventListener("click", this._onDocClick);
document.addEventListener("keydown", this._onKey);
this.refresh();
this._timer = setInterval(() => this.refresh(), POLL_MS);
}
disconnectedCallback() {
if (this._timer) clearInterval(this._timer);
this._timer = null;
document.removeEventListener("click", this._onDocClick);
document.removeEventListener("keydown", this._onKey);
}
get endpoint() {
return this.getAttribute("endpoint") || "/api/neighbors";
}
get current() {
return this.getAttribute("current") || "";
}
async refresh() {
try {
const res = await fetch(this.endpoint, {
headers: { Accept: "application/json" },
});
if (!res.ok) throw new Error(String(res.status));
const data = await res.json();
// `services` is the legacy alias some backends still emit.
this._neighbors = data.neighbors || data.services || [];
} catch {
// Silent: the menu is navigation, not a health indicator. Keep whatever
// we last had rather than blanking the list on one failed poll.
}
this._render();
}
_setOpen(open) {
this._open = open;
this._panel.hidden = !open;
this._btn.setAttribute("aria-expanded", String(open));
if (open) this.refresh();
}
_render() {
const panel = this._panel;
panel.textContent = "";
const seen = new Set();
const rows = [];
for (const n of this._neighbors) {
const name = n.name || "";
if (!name || seen.has(name)) continue;
seen.add(name);
rows.push(n);
}
// Always show the host itself, even before its own echo comes back.
if (this.current && !seen.has(this.current)) {
rows.push({ name: this.current, baseUrl: "" });
}
rows.sort((a, b) => String(a.name).localeCompare(String(b.name)));
if (rows.length === 0) {
const empty = document.createElement("div");
empty.className = "empty";
empty.textContent = "No services found";
panel.appendChild(empty);
return;
}
for (const n of rows) {
const isActive = n.name === this.current;
let url = n.baseUrl || "";
if (typeof this.rewriteUrl === "function") {
try {
url = this.rewriteUrl(n.name, url, n) || "";
} catch {
/* a bad hook must not blank the menu */
}
}
// A neighbour with no reachable URL is shown but not linked — better
// than a link that goes nowhere.
const el = document.createElement(isActive || !url ? "span" : "a");
el.className =
"item" + (isActive ? " active" : "") + (!isActive && !url ? " dead" : "");
if (el.tagName === "A") {
el.href = url;
el.setAttribute("role", "menuitem");
}
if (!isActive && !url) el.title = "No reachable URL announced";
const icon = document.createElement("span");
icon.className = "icon";
icon.textContent = ICONS[n.name] || DEFAULT_ICON;
const label = document.createElement("span");
label.textContent = titleCase(n.name);
el.append(icon, label);
if (n.status === "running" || isActive) {
const dot = document.createElement("span");
dot.className = "dot";
dot.title = n.status || "running";
el.appendChild(dot);
}
panel.appendChild(el);
}
}
}
if (!customElements.get("meta-service-menu")) {
customElements.define("meta-service-menu", MetaServiceMenu);
}
export default MetaServiceMenu;
-1
View File
@@ -10,7 +10,6 @@ set -e
# Create required directories
mkdir -p /meta-core/locks
mkdir -p /meta-core/db/redis
mkdir -p /meta-core/services
mkdir -p /meta-core/mounts/errors
mkdir -p /files
mkdir -p /var/log/supervisor
+7 -29
View File
@@ -2,9 +2,13 @@
# =============================================================================
# Leader Election Script for meta-core
#
# Uses flock for leader election. Only the leader runs the full application
# stack (supervisord with Redis, meta-core, nginx, rclone, mount-watcher).
# Followers remain dormant, periodically retrying to acquire the lock.
# The flock here is a MUTEX, not service discovery. It is the only thing
# stopping two containers that mount the same /meta-core volume from putting
# two Redis writers on one RDB/AOF. Sibling services no longer read anything
# from this directory — they locate meta-core over UDP (meta-discovery v1, see
# docs/project-architecture/service-discovery.md), so the kv-leader.info file
# this script used to publish is gone. Keep the lock; it costs nothing and the
# corruption it prevents is unrecoverable.
#
# Environment variables:
# META_CORE_PATH - Path to meta-core data directory (default: /meta-core)
@@ -20,7 +24,6 @@ META_CORE_HTTP_PORT="${META_CORE_HTTP_PORT:-9000}"
ELECTION_RETRY_SECS="${ELECTION_RETRY_SECS:-5}"
LOCK_FILE="${META_CORE_PATH}/locks/kv-leader.lock"
INFO_FILE="${META_CORE_PATH}/locks/kv-leader.info"
SUPERVISORD_CONF="/etc/supervisor/conf.d/supervisord.conf"
# Get the container's IP address
@@ -40,35 +43,13 @@ get_local_ip() {
echo "$ip"
}
# Write leader info file (plain text API URL)
write_leader_info() {
local ip=$(get_local_ip)
local api_url="http://${ip}:${META_CORE_HTTP_PORT}"
# Write atomically using temp file + rename
local temp_file="${INFO_FILE}.tmp"
echo "$api_url" > "$temp_file"
mv "$temp_file" "$INFO_FILE"
echo "[election] Wrote leader info: $api_url"
}
# Clean up leader info file on shutdown
cleanup() {
echo "[election] Cleaning up..."
rm -f "$INFO_FILE"
exit 0
}
# Set up signal handlers for graceful shutdown
trap cleanup SIGTERM SIGINT
# Ensure lock directory exists
mkdir -p "$(dirname "$LOCK_FILE")"
echo "[election] Starting leader election..."
echo "[election] Lock file: $LOCK_FILE"
echo "[election] Info file: $INFO_FILE"
# Open lock file for flock (file descriptor 200)
exec 200>"$LOCK_FILE"
@@ -79,9 +60,6 @@ while true; do
if flock -n 200; then
echo "[election] Acquired flock - transitioning to LEADER"
# Write leader info before starting services
write_leader_info
# exec supervisord - replaces this process
# When supervisord exits, the flock is released automatically
echo "[election] Starting supervisord..."
+4 -8
View File
@@ -123,14 +123,10 @@ http {
# =========================================================================
# Service discovery
# =========================================================================
location /services {
proxy_pass http://meta-core-api/services;
proxy_http_version 1.1;
proxy_set_header Host $host;
}
location /services/ {
proxy_pass http://meta-core-api/services/;
# meta-discovery v1 neighbours (supersedes /services, which stays
# while consumers migrate).
location /neighbors {
proxy_pass http://meta-core-api/neighbors;
proxy_http_version 1.1;
proxy_set_header Host $host;
}
+4
View File
@@ -301,6 +301,7 @@ CID-addressed (public, auth-bypassed at the perimeter)
Editor / KV / schema / snapshot
GET /api/metadata/hash-ids
GET /api/files/tuples
GET /api/metadata/list
POST /api/metadata/search
POST /api/metadata/batch
@@ -339,6 +340,9 @@ Watchers
Admin
POST /api/admin/migrate-dual-roots
POST /api/admin/sweep-literature-echoes (dry-run unless ?apply=true;
strips query echoes off domain=literature fileType=card records —
see storage/literature_echo_sweep.go)
WebDAV
/webdav/... (GET, PUT, DELETE, MKCOL, COPY, MOVE, PROPFIND)
+1
View File
@@ -18,4 +18,5 @@ require (
github.com/cespare/xxhash/v2 v2.2.0 // indirect
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
github.com/yuin/gopher-lua v1.1.0 // indirect
golang.org/x/sys v0.28.0 // indirect
)
+2
View File
@@ -34,3 +34,5 @@ golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ss
golang.org/x/net v0.21.0 h1:AQyQV4dYCvJ7vGmJyKki9+PBdyvhkSd8EIx/qb0AYv4=
golang.org/x/net v0.21.0/go.mod h1:bIjVDfnllIU7BJ2DNgfnXvpSvtn8VRwhlsaeUTyUS44=
golang.org/x/sys v0.0.0-20190204203706-41f3e6584952/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.28.0 h1:Fksou7UEQUWlKvIdsqzJmUmCX3cZuD2+P3XyyzwMhlA=
golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+75
View File
@@ -2,6 +2,7 @@ package api
import (
"net/http"
"strconv"
)
// handleMigrateDualRoots triggers the one-shot sweep that re-unifies
@@ -54,3 +55,77 @@ func (s *Server) handleMigrateDomainScreen(w http.ResponseWriter, r *http.Reques
"fixed": fixed,
})
}
// handleSweepLiteratureEchoes removes query echoes stored on literature cards:
// `anilistid` on a site card, and `workcid` / position keys on any card
// (meta-read docs/reading-model.md §7). See storage.SweepLiteratureEchoes for
// why chapters and anchor cards' ids are deliberately left alone.
//
// Dry-run unless `?apply=true`: unlike the migrations above this one deletes
// keys, so a bare call only reports what it would delete. An unparseable
// `apply` is a 400 rather than a silent dry run, so a typo cannot pass for a
// sweep that ran.
//
// Returns {success, dryRun, planned, fixed, byKey} — planned/fixed count
// records, byKey counts fields per key. Safe to call repeatedly: a swept
// record plans nothing on the next pass.
func (s *Server) handleSweepLiteratureEchoes(w http.ResponseWriter, r *http.Request) {
if !s.storage.IsConnected() {
writeError(w, http.StatusServiceUnavailable, "storage not connected")
return
}
apply := false
if v := r.URL.Query().Get("apply"); v != "" {
parsed, err := strconv.ParseBool(v)
if err != nil {
writeError(w, http.StatusBadRequest, "apply must be true or false")
return
}
apply = parsed
}
res, err := s.storage.SweepLiteratureEchoesWithTimeout(apply)
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
writeJSON(w, http.StatusOK, map[string]interface{}{
"success": true,
"dryRun": !apply,
"planned": res.Planned,
"fixed": res.Fixed,
"byKey": res.ByKey,
})
}
// handleMigrateUserCids triggers the one-shot sweep that builds
// `udl:idx:user:<uid>:cids` for User Data Layer records written before that
// index existed. See storage.UDLBackfillUserCids.
//
// The index is what makes whole-profile export a set read instead of a
// keyspace scan, and it is maintained by the upsert script from now on — so
// this only ever matters once, for data already on disk. Until it has run, a
// profile that predates the index exports as **empty**, which means a
// cross-instance recovery reports success having restored nothing.
//
// Returns {fixed: N} — cids added, not cids examined. Safe to call repeatedly:
// a second pass adds nothing.
func (s *Server) handleMigrateUserCids(w http.ResponseWriter, r *http.Request) {
if !s.storage.IsConnected() {
writeError(w, http.StatusServiceUnavailable, "storage not connected")
return
}
fixed, err := s.storage.UDLBackfillUserCids()
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
writeJSON(w, http.StatusOK, map[string]interface{}{
"success": true,
"fixed": fixed,
})
}
+93
View File
@@ -0,0 +1,93 @@
package api
import (
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
"github.com/alicebob/miniredis/v2"
"github.com/metazla/meta-core/internal/storage"
)
// newSweepTestServer is a Server over miniredis with one literature card
// carrying a copied `workcid`.
func newSweepTestServer(t *testing.T) *Server {
t.Helper()
mr := miniredis.RunT(t)
c := storage.NewClient("")
if err := c.Connect("redis://" + mr.Addr()); err != nil {
t.Fatalf("connect: %v", err)
}
t.Cleanup(func() { _ = c.Close() })
for k, v := range map[string]string{
"domain": "literature",
"fileType": "card",
"workcid": "bagdsaaaua5qw42lmnfzxi3lbnztwcorrga2tonzy",
} {
if err := c.SetProperty("card-1", k, v); err != nil {
t.Fatalf("seed %s: %v", k, err)
}
}
return &Server{storage: c}
}
type sweepResponse struct {
Success bool `json:"success"`
DryRun bool `json:"dryRun"`
Planned int `json:"planned"`
Fixed int `json:"fixed"`
ByKey map[string]int `json:"byKey"`
}
func postSweep(t *testing.T, s *Server, query string) (*httptest.ResponseRecorder, sweepResponse) {
t.Helper()
rr := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodPost, "/api/admin/sweep-literature-echoes"+query, nil)
s.handleSweepLiteratureEchoes(rr, req)
var body sweepResponse
if rr.Code == http.StatusOK {
if err := json.Unmarshal(rr.Body.Bytes(), &body); err != nil {
t.Fatalf("decode %s: %v", rr.Body.String(), err)
}
}
return rr, body
}
// A bare call must never delete: the default is the dry run.
func TestSweepLiteratureEchoes_DefaultsToDryRun(t *testing.T) {
s := newSweepTestServer(t)
rr, body := postSweep(t, s, "")
if rr.Code != http.StatusOK {
t.Fatalf("status = %d, body %s", rr.Code, rr.Body.String())
}
if !body.Success || !body.DryRun || body.Planned != 1 || body.Fixed != 0 || body.ByKey["workcid"] != 1 {
t.Errorf("dry run = %+v", body)
}
if v, _ := s.storage.GetProperty("card-1", "workcid"); v == "" {
t.Errorf("dry run deleted workcid")
}
rr, body = postSweep(t, s, "?apply=true")
if rr.Code != http.StatusOK {
t.Fatalf("status = %d, body %s", rr.Code, rr.Body.String())
}
if body.DryRun || body.Planned != 1 || body.Fixed != 1 || body.ByKey["workcid"] != 1 {
t.Errorf("apply = %+v", body)
}
if v, _ := s.storage.GetProperty("card-1", "workcid"); v != "" {
t.Errorf("apply left workcid = %q", v)
}
}
// A typo must not pass for a sweep that ran.
func TestSweepLiteratureEchoes_RejectsUnparseableApply(t *testing.T) {
s := newSweepTestServer(t)
rr, _ := postSweep(t, s, "?apply=yes-please")
if rr.Code != http.StatusBadRequest {
t.Errorf("status = %d, want 400", rr.Code)
}
}
+132
View File
@@ -0,0 +1,132 @@
package api
import (
"encoding/json"
"net/http"
"os"
"path"
"path/filepath"
"strings"
"github.com/metazla/meta-core/internal/watcher"
)
// registerFileRequest is the body of POST /api/files/register.
type registerFileRequest struct {
// Path relative to FILES_PATH, e.g. "plugin/subtitles/ab12….eng.srt".
Path string `json:"path"`
}
type registerFileResponse struct {
Root string `json:"root"`
CID string `json:"cid"`
Created bool `json:"created"`
Path string `json:"path"`
Size int64 `json:"size"`
}
// pluginRelPath normalises a FILES_PATH-relative path and refuses anything
// outside plugin/. Plugin output is the only tree a service may register this
// way: /files/watch is the user's library and belongs to the watcher.
func pluginRelPath(raw string) (string, bool) {
p := strings.TrimPrefix(filepath.ToSlash(strings.TrimSpace(raw)), "/")
if p == "" {
return "", false
}
for _, seg := range strings.Split(p, "/") {
if seg == ".." {
return "", false
}
}
clean := path.Clean(p)
if !strings.HasPrefix(clean, "plugin/") || clean == "plugin/" {
return "", false
}
return clean, true
}
// handleRegisterFile handles POST /api/files/register.
//
// Does for ONE file what the watcher does for every file it scans: midhash256
// it, mint a root carrying filePath/sizeByte/mtimeNano, register the
// cid:<midhash> alias, and append a file:events entry so meta-share's
// ipfs_seed chunks, seeds and provides it.
//
// It exists for cores that run with ENABLE_FILE_WATCHER=false (a meta-watch
// client box). Without it, bytes written under /files/plugin over WebDAV are
// never resolvable by CID and never reach the swarm. Idempotent: the same
// content at the same path returns the existing root with created=false.
func (s *Server) handleRegisterFile(w http.ResponseWriter, r *http.Request) {
if s.storage == nil || !s.storage.IsConnected() {
writeError(w, http.StatusServiceUnavailable, "storage not connected")
return
}
var req registerFileRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeError(w, http.StatusBadRequest, "invalid JSON body")
return
}
rel, ok := pluginRelPath(req.Path)
if !ok {
writeError(w, http.StatusBadRequest, "path must be a file under plugin/")
return
}
fullPath := filepath.Join(s.config.FilesPath, filepath.FromSlash(rel))
absFiles, _ := filepath.Abs(filepath.Join(s.config.FilesPath, "plugin"))
absFull, _ := filepath.Abs(fullPath)
if !strings.HasPrefix(absFull, absFiles+string(filepath.Separator)) {
writeError(w, http.StatusBadRequest, "path must be a file under plugin/")
return
}
info, err := os.Stat(fullPath)
if err != nil {
writeError(w, http.StatusNotFound, "file not found")
return
}
if info.IsDir() {
writeError(w, http.StatusBadRequest, "path is a directory, not a file")
return
}
midhash, err := watcher.ComputeMidHash256(fullPath, info.Size())
if err != nil {
writeError(w, http.StatusInternalServerError, "failed to hash file")
return
}
root, created, err := s.storage.RegisterFileTuple(fullPath, info.Size(), info.ModTime().UnixNano(), midhash)
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
if created {
ev := watcher.FileEvent{
Type: watcher.EventTypeAdd,
Path: rel,
Size: info.Size(),
MidHash256: midhash,
Timestamp: watcher.NowMS(),
}
if err := watcher.PublishEvent(s.storage, ev); err != nil {
// The record is written and resolvable; only seeding is late.
writeError(w, http.StatusInternalServerError, "registered, but file event not published: "+err.Error())
return
}
}
status := http.StatusOK
if created {
status = http.StatusCreated
}
writeJSON(w, status, registerFileResponse{
Root: root,
CID: midhash,
Created: created,
Path: rel,
Size: info.Size(),
})
}
+144
View File
@@ -0,0 +1,144 @@
package api
import (
"encoding/json"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"strings"
"testing"
"github.com/alicebob/miniredis/v2"
"github.com/metazla/meta-core/internal/config"
"github.com/metazla/meta-core/internal/storage"
)
// newRegisterTestServer wires handleRegisterFile to an in-memory Redis and a
// temp FILES_PATH. Built directly, not through NewServer, so no watcher runs —
// which is exactly the configuration this route exists for.
func newRegisterTestServer(t *testing.T) (*Server, string, *miniredis.Miniredis) {
t.Helper()
mr := miniredis.RunT(t)
c := storage.NewClient("")
if err := c.Connect("redis://" + mr.Addr()); err != nil {
t.Fatalf("connect to miniredis: %v", err)
}
t.Cleanup(func() { _ = c.Close() })
files := t.TempDir()
return &Server{storage: c, config: &config.Config{FilesPath: files}}, files, mr
}
func writeFixture(t *testing.T, files, rel, body string) string {
t.Helper()
full := filepath.Join(files, filepath.FromSlash(rel))
if err := os.MkdirAll(filepath.Dir(full), 0o755); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(full, []byte(body), 0o644); err != nil {
t.Fatal(err)
}
return full
}
func postRegister(t *testing.T, s *Server, path string) (int, registerFileResponse) {
t.Helper()
body, _ := json.Marshal(registerFileRequest{Path: path})
req := httptest.NewRequest(http.MethodPost, "/api/files/register", strings.NewReader(string(body)))
rr := httptest.NewRecorder()
s.handleRegisterFile(rr, req)
var resp registerFileResponse
_ = json.Unmarshal(rr.Body.Bytes(), &resp)
return rr.Code, resp
}
func streamLen(t *testing.T, mr *miniredis.Miniredis) int {
t.Helper()
entries, err := mr.Stream("file:events")
if err != nil {
return 0
}
return len(entries)
}
// The whole point of the route: with the watcher off, a file written under
// plugin/ becomes a record resolvable by its midhash, and meta-share hears
// about it on file:events.
func TestRegisterFile_MintsAliasAndPublishesEvent(t *testing.T) {
s, files, mr := newRegisterTestServer(t)
rel := "plugin/subtitles/ab12cd34.eng.srt"
full := writeFixture(t, files, rel, "1\n00:00:01,000 --> 00:00:02,000\nHello\n")
code, resp := postRegister(t, s, rel)
if code != http.StatusCreated || !resp.Created {
t.Fatalf("want 201 created, got %d %+v", code, resp)
}
if !strings.HasPrefix(resp.CID, "b") || resp.Root == "" {
t.Fatalf("want a multibase cid and a root, got %+v", resp)
}
root, err := s.storage.GetByCID(resp.CID)
if err != nil || root != resp.Root {
t.Fatalf("cid alias must resolve to the root: got %q err=%v, want %q", root, err, resp.Root)
}
if fp, _ := s.storage.GetProperty(resp.Root, "filePath"); fp != full {
t.Fatalf("filePath = %q, want %q", fp, full)
}
entries, err := mr.Stream("file:events")
if err != nil || len(entries) != 1 {
t.Fatalf("want exactly one file:events entry, got %d err=%v", len(entries), err)
}
vals := strings.Join(entries[0].Values, "|")
if !strings.Contains(vals, "midhash256|"+resp.CID) || !strings.Contains(vals, "path|"+rel) {
t.Fatalf("event must carry the midhash and the relative path, got %q", vals)
}
}
// Re-uploading the same subtitle is a no-op: same root, no second event (which
// would make meta-share re-chunk and re-provide the file).
func TestRegisterFile_IsIdempotent(t *testing.T) {
s, files, mr := newRegisterTestServer(t)
rel := "plugin/subtitles/same.eng.srt"
writeFixture(t, files, rel, "same bytes")
_, first := postRegister(t, s, rel)
code, second := postRegister(t, s, rel)
if code != http.StatusOK || second.Created {
t.Fatalf("second register must be 200 created=false, got %d %+v", code, second)
}
if second.Root != first.Root || second.CID != first.CID {
t.Fatalf("second register must return the same record: %+v vs %+v", second, first)
}
if n := streamLen(t, mr); n != 1 {
t.Fatalf("an idempotent re-register must not publish again, stream has %d", n)
}
}
// Only plugin output may be registered here. /files/watch is the user's
// library and belongs to the watcher; traversal must not escape either.
func TestRegisterFile_RejectsPathsOutsidePlugin(t *testing.T) {
s, files, _ := newRegisterTestServer(t)
writeFixture(t, files, "watch/movie.mkv", "x")
for _, p := range []string{
"", "plugin", "plugin/", "watch/movie.mkv", "/watch/movie.mkv",
"plugin/../watch/movie.mkv", "../etc/passwd", "plugin/../../etc/passwd",
} {
if code, _ := postRegister(t, s, p); code != http.StatusBadRequest {
t.Errorf("path %q: want 400, got %d", p, code)
}
}
}
func TestRegisterFile_MissingFileAndDirectory(t *testing.T) {
s, files, _ := newRegisterTestServer(t)
if code, _ := postRegister(t, s, "plugin/subtitles/nope.srt"); code != http.StatusNotFound {
t.Errorf("missing file: want 404, got %d", code)
}
if err := os.MkdirAll(filepath.Join(files, "plugin", "subtitles", "dir"), 0o755); err != nil {
t.Fatal(err)
}
if code, _ := postRegister(t, s, "plugin/subtitles/dir"); code != http.StatusBadRequest {
t.Errorf("directory: want 400, got %d", code)
}
}
+74
View File
@@ -0,0 +1,74 @@
package api
import "net/http"
// fileTupleJSON is one file-backed root on the wire.
type fileTupleJSON struct {
HashID string `json:"hashId"`
FilePath string `json:"filePath"`
SizeByte int64 `json:"sizeByte"`
MtimeNano int64 `json:"mtimeNano"`
}
// FileTuplesResponse is the body of GET /api/files/tuples.
//
// Files is a pointer so the summary form can omit it while the full form
// always carries it — an empty library is `"files": []`, never an absent key,
// which a client could mistake for a meta-core that lacks the rows.
type FileTuplesResponse struct {
Files *[]fileTupleJSON `json:"files,omitempty"`
Count int `json:"count"`
TotalSize int64 `json:"totalSize"`
}
// handleGetFileTuples handles GET /api/files/tuples.
//
// Lists the FILE-BACKED roots — records carrying filePath, sizeByte and
// mtimeNano, i.e. what the watcher or /api/files/register minted — with their
// sizes, plus the count and total. Gateway / cid-rooted records carry no file
// and are not included; neither is a locator record that only merged a cache
// file's metadata (it has no mtimeNano).
//
// It exists so a client never walks every hashId to answer "how many files,
// how big": GetTuplesForAllFiles reads the three fields for every root in
// chunked MGETs server-side (~250 round-trips for 82k roots, no SCAN), where
// the client-side equivalent was one HTTP request per record.
//
// ?summary=1 omits the rows and returns only {count, totalSize}.
func (s *Server) handleGetFileTuples(w http.ResponseWriter, r *http.Request) {
if !s.storage.IsConnected() {
writeError(w, http.StatusServiceUnavailable, "storage not connected")
return
}
tuples, err := s.storage.GetTuplesForAllFiles()
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
summary := r.URL.Query().Get("summary")
withRows := summary == "" || summary == "0" || summary == "false"
resp := FileTuplesResponse{Count: len(tuples)}
var rows []fileTupleJSON
if withRows {
rows = make([]fileTupleJSON, 0, len(tuples))
}
for _, t := range tuples {
resp.TotalSize += t.Size
if withRows {
rows = append(rows, fileTupleJSON{
HashID: t.HashID,
FilePath: t.FilePath,
SizeByte: t.Size,
MtimeNano: t.MtimeNano,
})
}
}
if withRows {
resp.Files = &rows
}
writeJSON(w, http.StatusOK, resp)
}
+114
View File
@@ -0,0 +1,114 @@
package api
import (
"encoding/json"
"net/http"
"net/http/httptest"
"sort"
"testing"
"github.com/alicebob/miniredis/v2"
"github.com/metazla/meta-core/internal/storage"
)
func newFileTuplesTestServer(t *testing.T) (*Server, *storage.Client) {
t.Helper()
mr := miniredis.RunT(t)
c := storage.NewClient("")
if err := c.Connect("redis://" + mr.Addr()); err != nil {
t.Fatalf("connect to miniredis: %v", err)
}
t.Cleanup(func() { _ = c.Close() })
return &Server{storage: c}, c
}
func getFileTuples(t *testing.T, s *Server, query string) (int, FileTuplesResponse) {
t.Helper()
req := httptest.NewRequest(http.MethodGet, "/api/files/tuples"+query, nil)
rr := httptest.NewRecorder()
s.handleGetFileTuples(rr, req)
var resp FileTuplesResponse
_ = json.Unmarshal(rr.Body.Bytes(), &resp)
return rr.Code, resp
}
func mustMint(t *testing.T, c *storage.Client, path string, size, mtime int64) string {
t.Helper()
id, err := c.Mint(path, size, mtime)
if err != nil {
t.Fatalf("mint %s: %v", path, err)
}
return id
}
// Only records that are actually files count: a gateway record and a record
// that merely carries a path + size (no mtimeNano) are both left out.
func TestFileTuples_OnlyFileBackedRoots(t *testing.T) {
s, c := newFileTuplesTestServer(t)
a := mustMint(t, c, "/files/watch/a.mkv", 1000, 1)
b := mustMint(t, c, "/files/watch/b.mkv", 2500, 2)
if err := c.SetMetadataFlat("bagcsgatewayrecord", map[string]string{"title": "Some Release"}); err != nil {
t.Fatal(err)
}
if err := c.SetMetadataFlat("01partialroot", map[string]string{"filePath": "/files/cache/x.mkv", "sizeByte": "999"}); err != nil {
t.Fatal(err)
}
code, resp := getFileTuples(t, s, "")
if code != http.StatusOK {
t.Fatalf("want 200, got %d", code)
}
if resp.Count != 2 || resp.TotalSize != 3500 {
t.Fatalf("want count 2 / totalSize 3500, got %d / %d", resp.Count, resp.TotalSize)
}
if resp.Files == nil || len(*resp.Files) != 2 {
t.Fatalf("want 2 rows, got %+v", resp.Files)
}
got := []string{(*resp.Files)[0].HashID, (*resp.Files)[1].HashID}
want := []string{a, b}
sort.Strings(got)
sort.Strings(want)
if got[0] != want[0] || got[1] != want[1] {
t.Fatalf("want roots %v, got %v", want, got)
}
}
func TestFileTuples_SummaryOmitsRows(t *testing.T) {
s, c := newFileTuplesTestServer(t)
mustMint(t, c, "/files/watch/a.mkv", 1000, 1)
mustMint(t, c, "/files/watch/b.mkv", 2500, 2)
code, resp := getFileTuples(t, s, "?summary=1")
if code != http.StatusOK {
t.Fatalf("want 200, got %d", code)
}
if resp.Files != nil {
t.Fatalf("summary must omit rows, got %d", len(*resp.Files))
}
if resp.Count != 2 || resp.TotalSize != 3500 {
t.Fatalf("want count 2 / totalSize 3500, got %d / %d", resp.Count, resp.TotalSize)
}
}
// An empty library still carries "files": [] — clients treat an absent key as
// "this meta-core cannot list files", so it must not disappear when empty.
func TestFileTuples_EmptyLibraryKeepsFilesKey(t *testing.T) {
s, _ := newFileTuplesTestServer(t)
code, resp := getFileTuples(t, s, "")
if code != http.StatusOK {
t.Fatalf("want 200, got %d", code)
}
if resp.Files == nil || len(*resp.Files) != 0 || resp.Count != 0 {
t.Fatalf("want an empty files array and count 0, got %+v", resp)
}
}
func TestFileTuples_StorageNotConnected(t *testing.T) {
s := &Server{storage: storage.NewClient("")}
code, _ := getFileTuples(t, s, "")
if code != http.StatusServiceUnavailable {
t.Fatalf("want 503, got %d", code)
}
}
+26 -47
View File
@@ -14,6 +14,7 @@ import (
"github.com/gorilla/mux"
"github.com/metazla/meta-core/internal/leader"
"github.com/metazla/meta-core/internal/meshdisco"
)
// contentTypeByExt maps file extensions to MIME types
@@ -547,62 +548,40 @@ func (s *Server) handleGetFileByCID(w http.ResponseWriter, r *http.Request) {
http.ServeContent(w, r, fileInfo.Name(), fileInfo.ModTime(), file)
}
// handleListServices handles GET /services and GET /api/services
// Supports optional ?current=<service-name> query parameter
func (s *Server) handleListServices(w http.ResponseWriter, r *http.Request) {
services, err := s.discovery.DiscoverAll()
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
// handleListNeighbors handles GET /neighbors and GET /api/neighbors.
//
// meta-discovery v1. The list is built from UDP announces held in memory, not
// from files under /meta-core/services, so it works across stacks that share a
// network but no volume. Instances of the same service are collapsed to the
// most recent, which preserves the one-row-per-service shape the old
// /api/services had. `?all=1` returns every instance, for debugging.
func (s *Server) handleListNeighbors(w http.ResponseWriter, r *http.Request) {
response := map[string]interface{}{
"services": services,
"count": len(services),
}
// Add 'current' field if query parameter is provided
current := r.URL.Query().Get("current")
if current != "" {
response["current"] = current
"current": s.config.ServiceName,
"neighbors": []interface{}{},
"count": 0,
}
if s.mesh == nil {
// Discovery off or failed to bind. Report empty rather than 503 —
// the nav is decoration and callers already treat [] as "hide".
response["enabled"] = false
writeJSON(w, http.StatusOK, response)
}
// handleGetService handles GET /services/{name}
func (s *Server) handleGetService(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r)
name := vars["name"]
if name == "" {
writeError(w, http.StatusBadRequest, "service name is required")
return
}
service, err := s.discovery.Discover(name)
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
var list []meshdisco.Neighbor
if r.URL.Query().Get("all") != "" {
list = s.mesh.Neighbors()
} else {
list = s.mesh.NeighborsByName()
}
if service == nil {
writeError(w, http.StatusNotFound, "service not found")
return
}
writeJSON(w, http.StatusOK, service)
}
// handleCleanupStats handles GET /services/cleanup/stats
func (s *Server) handleCleanupStats(w http.ResponseWriter, r *http.Request) {
if s.cleaner == nil {
writeError(w, http.StatusServiceUnavailable, "cleaner not initialized")
return
}
stats := s.cleaner.Stats()
writeJSON(w, http.StatusOK, stats)
response["enabled"] = true
response["neighbors"] = list
response["count"] = len(list)
response["self"] = s.mesh.Self()
writeJSON(w, http.StatusOK, response)
}
// writeJSON writes a JSON response
+30 -13
View File
@@ -9,10 +9,10 @@ import (
"github.com/gorilla/mux"
"github.com/metazla/meta-core/internal/config"
"github.com/metazla/meta-core/internal/discovery"
"github.com/metazla/meta-core/internal/events"
"github.com/metazla/meta-core/internal/identity"
"github.com/metazla/meta-core/internal/leader"
"github.com/metazla/meta-core/internal/meshdisco"
"github.com/metazla/meta-core/internal/mounts"
"github.com/metazla/meta-core/internal/schema"
"github.com/metazla/meta-core/internal/storage"
@@ -25,8 +25,6 @@ import (
type Server struct {
config *config.Config
leaderProvider *leader.LeaderInfoProvider
discovery *discovery.Service
cleaner *discovery.Cleaner
storage *storage.Client
mountsManager *mounts.Manager
mountsHandlers *mounts.Handlers
@@ -45,23 +43,30 @@ type Server struct {
// challenges backs proof-of-possession on reveal/delete. In-process, so a
// restart invalidates the ones in flight — see identity/challenge.go.
challenges *identity.ChallengeStore
// mesh is the UDP discovery node (meta-discovery v1). nil when discovery
// is disabled or failed to bind, in which case /api/neighbors reports an
// empty list rather than erroring — a dashboard nav is decoration and
// must not take the API down with it.
mesh *meshdisco.Node
router *mux.Router
server *http.Server
}
// SetMeshNode attaches the UDP discovery node. Separate from NewServer so the
// server's constructor signature (and its tests) stay untouched.
func (s *Server) SetMeshNode(n *meshdisco.Node) {
s.mesh = n
}
// NewServer creates a new API server
func NewServer(
cfg *config.Config,
leaderProvider *leader.LeaderInfoProvider,
disc *discovery.Service,
cleaner *discovery.Cleaner,
stor *storage.Client,
) *Server {
s := &Server{
config: cfg,
leaderProvider: leaderProvider,
discovery: disc,
cleaner: cleaner,
storage: stor,
search: &searchIndex{},
router: mux.NewRouter(),
@@ -216,6 +221,8 @@ func (s *Server) setupRoutes() {
s.router.HandleFunc("/api/udl/records", s.handleUDLRecordsPut).Methods("PUT")
s.router.HandleFunc("/api/udl/user/{uid}/key/{key}", s.handleUDLUserKey).Methods("GET")
s.router.HandleFunc("/api/udl/user/{uid}/cid/{cid}", s.handleUDLUserCid).Methods("GET")
s.router.HandleFunc("/api/udl/user/{uid}/all", s.handleUDLUserAll).Methods("GET")
s.router.HandleFunc("/api/udl/user/{uid}/count", s.handleUDLUserCount).Methods("GET")
s.router.HandleFunc("/api/udl/cid/{cid}/users", s.handleUDLCidUsers).Methods("GET")
s.router.HandleFunc("/api/udl/cid/{cid}/aggregate", s.handleUDLAggregate).Methods("GET")
s.router.HandleFunc("/api/udl/users/stats", s.handleUDLUserStats).Methods("GET")
@@ -243,6 +250,14 @@ func (s *Server) setupRoutes() {
s.router.HandleFunc("/api/file/{cid}/info", s.handleGetFileByCIDInfo).Methods("GET")
s.router.HandleFunc("/file/{cid}", s.handleGetFileByCID).Methods("GET")
s.router.HandleFunc("/file/cid", s.handleComputeFileCID).Methods("POST")
// Register one file under /files/plugin by midhash — the watcher's
// mint + alias + file:events path for a single file. Registered
// unconditionally: it exists for boxes whose watcher is OFF.
s.router.HandleFunc("/api/files/register", s.handleRegisterFile).Methods("POST")
// File-backed roots (filePath+sizeByte+mtimeNano) with sizes, batched
// server-side, so clients answer "how many files, how big" without one
// read per hashId. ?summary=1 returns only {count, totalSize}.
s.router.HandleFunc("/api/files/tuples", s.handleGetFileTuples).Methods("GET")
// Public metadata-by-CID resolver. Reverse-index lookup → full document.
// Auth-bypassed alongside /api/file/{cid} so meta-share peers can query
@@ -256,6 +271,10 @@ func (s *Server) setupRoutes() {
// One-shot vocabulary sweep: domain film|tv → screen, workForm backfill
// (METADATA_KEYS.md §14.17). Idempotent.
s.router.HandleFunc("/api/admin/migrate-domain-screen", s.handleMigrateDomainScreen).Methods("POST")
s.router.HandleFunc("/api/admin/migrate-user-cids", s.handleMigrateUserCids).Methods("POST")
// One-shot sweep of query echoes on literature cards (meta-read
// reading-model.md §7). Dry-run unless ?apply=true. Idempotent.
s.router.HandleFunc("/api/admin/sweep-literature-echoes", s.handleSweepLiteratureEchoes).Methods("POST")
// SSE event streams. Mediated mirror of the Redis Streams; consumers
// reach them over HTTP instead of touching Redis directly. Auth-bypass
@@ -264,12 +283,10 @@ func (s *Server) setupRoutes() {
s.router.HandleFunc("/api/events/files", s.handleEventsFiles).Methods("GET")
s.router.HandleFunc("/api/events/meta", s.handleEventsMeta).Methods("GET")
// Service discovery
// Both /services and /api/services are supported for consistency with other services
s.router.HandleFunc("/services", s.handleListServices).Methods("GET")
s.router.HandleFunc("/api/services", s.handleListServices).Methods("GET")
s.router.HandleFunc("/services/cleanup/stats", s.handleCleanupStats).Methods("GET")
s.router.HandleFunc("/services/{name}", s.handleGetService).Methods("GET")
// meta-discovery v1: neighbours heard over UDP. Supersedes /api/services,
// which is still served above while consumers migrate.
s.router.HandleFunc("/neighbors", s.handleListNeighbors).Methods("GET")
s.router.HandleFunc("/api/neighbors", s.handleListNeighbors).Methods("GET")
// Mount management routes (if manager initialized)
if s.mountsHandlers != nil {
+70
View File
@@ -13,6 +13,7 @@ import (
"encoding/json"
"math"
"net/http"
"strconv"
"strings"
"github.com/gorilla/mux"
@@ -271,6 +272,75 @@ func (s *Server) handleUDLUserCid(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, map[string]interface{}{"entries": entries})
}
// handleUDLUserAll handles GET /api/udl/user/{uid}/all?after=<cid>&limit=<n>
// — every record a uid owns, for cross-instance profile recovery.
//
// The other user-scoped reads answer a question with one coordinate already
// known. Recovery knows only the uid: a box that has just imported a secret key
// has no idea which cids or keys the profile ever touched, so it needs the
// whole set or it gets nothing.
//
// Paged on a cid boundary — `nextCid` empty means the export is complete. The
// response is opaque signed records, exactly as stored; whoever consumes it is
// responsible for verifying each one before trusting it (meta-core has never
// held the account key and cannot verify on anyone's behalf).
func (s *Server) handleUDLUserAll(w http.ResponseWriter, r *http.Request) {
if !s.storage.IsConnected() {
writeError(w, http.StatusServiceUnavailable, "storage not connected")
return
}
uid := mux.Vars(r)["uid"]
if uid == "" {
writeError(w, http.StatusBadRequest, "uid is required")
return
}
limit := maxUDLBatch
if v := r.URL.Query().Get("limit"); v != "" {
n, err := strconv.Atoi(v)
if err != nil || n <= 0 {
writeError(w, http.StatusBadRequest, "limit must be a positive integer")
return
}
if n < limit {
limit = n
}
}
entries, nextCid, err := s.storage.UDLAllForUser(uid, r.URL.Query().Get("after"), limit)
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
writeJSON(w, http.StatusOK, map[string]interface{}{
"entries": entries,
"nextCid": nextCid,
})
}
// handleUDLUserCount handles GET /api/udl/user/{uid}/count — how many cids a
// uid holds records for.
//
// Exists so a gateway mirror can enforce a per-profile quota without walking
// anything: 32 random bytes are a valid uid, so an unbounded mirror is a
// disk-fill DoS that needs no account. A quota check that itself cost a scan
// would just move the denial of service one layer up.
func (s *Server) handleUDLUserCount(w http.ResponseWriter, r *http.Request) {
if !s.storage.IsConnected() {
writeError(w, http.StatusServiceUnavailable, "storage not connected")
return
}
uid := mux.Vars(r)["uid"]
if uid == "" {
writeError(w, http.StatusBadRequest, "uid is required")
return
}
n, err := s.storage.UDLCountUserCids(uid)
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
writeJSON(w, http.StatusOK, map[string]interface{}{"cids": n})
}
// handleUDLCidUsers handles GET /api/udl/cid/{cid}/users (uids_for_cid).
func (s *Server) handleUDLCidUsers(w http.ResponseWriter, r *http.Request) {
if !s.storage.IsConnected() {
+92
View File
@@ -0,0 +1,92 @@
package cid
import (
"encoding/base32"
"encoding/binary"
"fmt"
"strings"
"unicode/utf8"
)
// DecodeCard recovers the (source, id) pair a card locator (CodeCard, 0x1007)
// wraps. Wire form:
//
// 0x01 ‖ varint(0x1007) ‖ 0x00 (identity mh) ‖ varint(len(digest)) ‖ digest
// digest = varint(len(source)) ‖ source ‖ id (both UTF-8)
//
// Decode stops at the multihash code because ranking needs nothing past it;
// reading *which work* a card addresses needs the digest too, so this is the
// one place in Go that walks the whole thing.
//
// ⚠ Mirror of meta-read `card_cid::decode` and of meta-feeder-sdk
// `compute_card_cid`. The vectors in card_test.go are pinned literals copied
// from meta-read, not round-trips, so a drift on either side fails a test
// instead of silently reading a different work out of the same CID.
//
// Identity multihash only: a 0x1007 CID carrying any other multihash is not
// something anyone mints, and reading its digest as (source, id) would be
// reading hash bytes as text. Bytes past the declared digest length are
// ignored, as meta-read does.
func DecodeCard(cidStr string) (source, id string, err error) {
if len(cidStr) < 2 || cidStr[0] != 'b' {
return "", "", fmt.Errorf("cid %q: not a multibase base32 CIDv1", cidStr)
}
raw, err := base32.StdEncoding.WithPadding(base32.NoPadding).DecodeString(strings.ToUpper(cidStr[1:]))
if err != nil {
return "", "", fmt.Errorf("cid %q: base32 decode: %w", cidStr, err)
}
next := func(buf []byte, what string) (uint64, []byte, error) {
v, n := binary.Uvarint(buf)
if n <= 0 {
return 0, nil, fmt.Errorf("cid %q: bad %s varint", cidStr, what)
}
return v, buf[n:], nil
}
version, raw, err := next(raw, "version")
if err != nil {
return "", "", err
}
if version != 1 {
return "", "", fmt.Errorf("cid %q: unsupported CID version %d", cidStr, version)
}
codec, raw, err := next(raw, "codec")
if err != nil {
return "", "", err
}
if codec != CodeCard {
return "", "", fmt.Errorf("cid %q: codec 0x%x is not a card locator", cidStr, codec)
}
mh, raw, err := next(raw, "multihash-code")
if err != nil {
return "", "", err
}
if mh != 0 {
return "", "", fmt.Errorf("cid %q: card locator with non-identity multihash 0x%x", cidStr, mh)
}
digestLen, raw, err := next(raw, "digest-length")
if err != nil {
return "", "", err
}
if digestLen > uint64(len(raw)) {
return "", "", fmt.Errorf("cid %q: digest truncated", cidStr)
}
digest := raw[:digestLen]
sourceLen, rest, err := next(digest, "source-length")
if err != nil {
return "", "", err
}
if sourceLen > uint64(len(rest)) {
return "", "", fmt.Errorf("cid %q: source truncated", cidStr)
}
source, id = string(rest[:sourceLen]), string(rest[sourceLen:])
if source == "" || id == "" {
return "", "", fmt.Errorf("cid %q: empty source or id", cidStr)
}
if !utf8.ValidString(source) || !utf8.ValidString(id) {
return "", "", fmt.Errorf("cid %q: source or id is not UTF-8", cidStr)
}
return source, id, nil
}
+71
View File
@@ -0,0 +1,71 @@
package cid
import (
"encoding/base32"
"encoding/binary"
"strings"
"testing"
)
// Pinned from meta-read `src/card_cid.rs` / meta-feeder-sdk `compute_card_cid`
// — literals, not round-trips (see DecodeCard).
const (
goldenAnilistCard = "bagdsaaaua5qw42lmnfzxi3lbnztwcorrga2tonzy"
goldenMangadexCard = "bagdsaabybrwwc3thmfsgk6bon5zgol3nmfxgoyjpmiygenzsgftgmlldgm4dqljugq4dmllbmeygmlldgjrdaytcgmzdcnjrgi"
)
// wireCID builds a CIDv1 from raw parts so the negative cases below can vary
// one field at a time.
func wireCID(codec, mh uint64, digest []byte) string {
buf := []byte{0x01}
buf = binary.AppendUvarint(buf, codec)
buf = binary.AppendUvarint(buf, mh)
buf = binary.AppendUvarint(buf, uint64(len(digest)))
buf = append(buf, digest...)
return "b" + strings.ToLower(base32.StdEncoding.WithPadding(base32.NoPadding).EncodeToString(buf))
}
func cardDigest(source, id string) []byte {
d := binary.AppendUvarint(nil, uint64(len(source)))
d = append(d, source...)
return append(d, id...)
}
func TestDecodeCard_GoldenVectors(t *testing.T) {
for _, tc := range []struct {
cid, source, id string
}{
{goldenAnilistCard, "anilist", "manga:105778"},
{goldenMangadexCard, "mangadex.org", "/manga/b0b721ff-c388-4486-aa0f-c2b0bb321512"},
} {
source, id, err := DecodeCard(tc.cid)
if err != nil {
t.Fatalf("DecodeCard(%s): %v", tc.cid, err)
}
if source != tc.source || id != tc.id {
t.Errorf("DecodeCard(%s) = (%q, %q), want (%q, %q)", tc.cid, source, id, tc.source, tc.id)
}
// The test encoder must agree with the pinned literal, or the negative
// cases below are testing a different wire format.
if got := wireCID(CodeCard, 0, cardDigest(tc.source, tc.id)); got != tc.cid {
t.Errorf("wireCID(%q, %q) = %s, want %s", tc.source, tc.id, got, tc.cid)
}
}
}
func TestDecodeCard_RejectsEverythingElse(t *testing.T) {
for name, s := range map[string]string{
"url locator codec": wireCID(CodeURL, 0, cardDigest("anilist", "1")),
"non-identity multihash": wireCID(CodeCard, CodeSha256, cardDigest("anilist", "1")),
"empty id": wireCID(CodeCard, 0, cardDigest("anilist", "")),
"source past digest": wireCID(CodeCard, 0, []byte{0x20, 'a'}),
"not a CID": "title:berserk",
"CIDv0": "QmYwAPJzv5CZsnA625s3Xf2nemtYgPpHdWEz79ojWnPbdG",
"truncated": goldenAnilistCard[:12],
"empty": "",
} {
if source, id, err := DecodeCard(s); err == nil {
t.Errorf("%s: DecodeCard(%q) = (%q, %q), want error", name, s, source, id)
}
}
}
+6
View File
@@ -128,6 +128,12 @@
"rank": 5,
"note": "codec 0x1009 + IDENTITY multihash 0x00, digest = the verbatim UTF-8 URL. WIRE-IDENTICAL to url (0x1006) apart from the codec, and that difference is the entire contract: 0x1006 means 'fetch these bytes once and seed them', 0x1009 means 'open this in a browser, there are no bytes for us'. A decoder that matched on SHAPE rather than on the codec slot would see an identity multihash wrapping an http(s) URL, take the 0x1006 path, fetch a watch/Bandcamp page, chunk it, seed the HTML and link that content CID onto the record AS THE FILE — the MSR1 poisoning shape, except permanent and replicated. Same tier/reasoning as the rest of the locator family."
},
{
"name": "provider-file-locator",
"cid": "bagfcaaa2bvxxazloon2we5djorwgk43gnfwgkorxga3dcobtgq",
"rank": 5,
"note": "codec 0x100A + IDENTITY multihash 0x00, digest = varint(len(source)) || source || id - here (\"opensubtitles\",\"file:7061834\"). The GENERIC fetchable locator: one codec for every provider whose bytes are fetched the same way; the provider is inside the digest. Framing byte-identical to card (0x1007), but the byte path RESOLVES it (via source's fetch tier) instead of 404ing. Same codec-slot rule as the rest of the family: match on the CODEC, never the mh (identity 0x00 would fall through and score 0). Tier 5: once the fetched bytes' sha2-256 is linked onto the record it wins the canonical election with no addressing change."
},
{
"name": "sha1",
"cid": "baeircff3eqvxxaqqot2lt242stkdtk3gq4c2vra",
+9 -1
View File
@@ -75,6 +75,14 @@ const (
CodeYtVideo = 0x1008 // custom — identity-multihash CID wrapping a YouTube (kind, id)
CodeExtPlay = 0x1009 // custom — identity-multihash CID wrapping a URL to OPEN, never fetch
// CodeProviderFile is the FETCHABLE twin of CodeExtPlay: a provider-held
// file addressed as varint(len(source))‖source‖id — ("opensubtitles",
// "file:7061834"). Framing is byte-identical to CodeCard; only the byte
// path differs (it redeems the id through that source's feeder instead of
// 404ing). Same codec-slot rule and same tier: once the fetched bytes'
// sha2-256 is linked onto the record it wins with no addressing change.
CodeProviderFile = 0x100A // custom — identity-multihash CID wrapping a (source, id) provider file
// CodeNzbPosting is the INVERSE trap of the two above: its multihash is a
// REAL sha2-256 digest (over the normalised, sorted article Message-ID
// set of a Usenet posting we scanned ourselves — never a locator, no
@@ -183,7 +191,7 @@ func Rank(cidStr string) int {
// Opaque locators. Codec-only: their multihash is identity (0x00), so
// there is nothing to match on the mh side.
if codec == CodeNzbRelease || codec == CodeURL || codec == CodeCard ||
codec == CodeYtVideo || codec == CodeExtPlay {
codec == CodeYtVideo || codec == CodeExtPlay || codec == CodeProviderFile {
return RankLocator
}
// nzb-posting: codec-only, checked BEFORE the mh switch below — its
+15 -11
View File
@@ -40,6 +40,14 @@ type Config struct {
// Mount configuration
MountsDir string // Path to mounts configuration (default: /meta-core/mounts)
// UDP service discovery (meta-discovery v1). Replaces the file-based
// registry that used to live under /meta-core/services; see
// docs/project-architecture/service-discovery.md.
EnableUDPDiscovery bool // Announce + listen on the multicast group (default: true)
DiscoveryGroup string // IPv4 multicast group (default: 239.255.77.1)
DiscoveryPort int // UDP port (default: 9399)
DiscoveryIntervalMS int // Unsolicited announce interval in ms (default: 10000)
}
// Load creates a Config from environment variables
@@ -63,6 +71,10 @@ func Load() *Config {
WatchIntervalMS: getEnvInt("WATCH_INTERVAL_MS", 1000),
DebounceMS: getEnvInt("DEBOUNCE_MS", 30000),
EnableFileWatcher: getEnvBool("ENABLE_FILE_WATCHER", true),
EnableUDPDiscovery: getEnvBool("ENABLE_UDP_DISCOVERY", true),
DiscoveryGroup: getEnv("DISCOVERY_GROUP", "239.255.77.1"),
DiscoveryPort: getEnvInt("DISCOVERY_PORT", 9399),
DiscoveryIntervalMS: getEnvInt("DISCOVERY_INTERVAL_MS", 10000),
}
// Set mounts directory
@@ -71,26 +83,18 @@ func Load() *Config {
return cfg
}
// LockFilePath returns the path to the leader lock file
// LockFilePath returns the path to the Redis-volume mutex. Held by
// docker/leader-election.sh; nothing reads it, and it is NOT service
// discovery — see docs/project-architecture/service-discovery.md.
func (c *Config) LockFilePath() string {
return c.MetaCorePath + "/locks/kv-leader.lock"
}
// InfoFilePath returns the path to the leader info file
func (c *Config) InfoFilePath() string {
return c.MetaCorePath + "/locks/kv-leader.info"
}
// RedisDataDir returns the path to Redis data directory
func (c *Config) RedisDataDir() string {
return c.MetaCorePath + "/db/redis"
}
// ServicesDir returns the path to services directory
func (c *Config) ServicesDir() string {
return c.MetaCorePath + "/services"
}
// MountsFilePath returns the path to the mounts configuration file
func (c *Config) MountsFilePath() string {
return c.MountsDir + "/mounts.json"
-196
View File
@@ -1,196 +0,0 @@
package discovery
import (
"encoding/json"
"log"
"os"
"path/filepath"
"sync"
"time"
"github.com/metazla/meta-core/internal/config"
)
// CleanupStats holds statistics about the cleanup process
type CleanupStats struct {
LastRunTime string `json:"lastRunTime,omitempty"`
ServicesRemoved int `json:"servicesRemoved"`
TotalRuns int `json:"totalRuns"`
Errors int `json:"errors"`
}
// Cleaner handles periodic cleanup of dead service registration files
type Cleaner struct {
config *config.Config
servicesDir string
ownFile string // Our own service file (never delete)
stats CleanupStats
stopChan chan struct{}
wg sync.WaitGroup
mu sync.RWMutex
}
// NewCleaner creates a new service cleaner
func NewCleaner(cfg *config.Config) *Cleaner {
hostname, _ := os.Hostname()
return &Cleaner{
config: cfg,
servicesDir: cfg.ServicesDir(),
ownFile: filepath.Join(cfg.ServicesDir(), cfg.ServiceName+"-"+hostname+".json"),
stopChan: make(chan struct{}),
}
}
// Start begins the cleanup loop
func (c *Cleaner) Start() error {
log.Printf("[Cleaner] Starting dead service cleanup (interval: %dms, threshold: %dms)",
c.config.CleanupIntervalMS, c.config.DeadServiceThresholdMS)
c.wg.Add(1)
go c.cleanupLoop()
return nil
}
// Stop stops the cleanup loop
func (c *Cleaner) Stop() error {
log.Println("[Cleaner] Stopping dead service cleanup...")
close(c.stopChan)
c.wg.Wait()
return nil
}
// cleanupLoop runs the periodic cleanup
func (c *Cleaner) cleanupLoop() {
defer c.wg.Done()
// Run an initial cleanup shortly after starting
initialDelay := time.NewTimer(30 * time.Second)
select {
case <-c.stopChan:
initialDelay.Stop()
return
case <-initialDelay.C:
c.runCleanup()
}
ticker := time.NewTicker(time.Duration(c.config.CleanupIntervalMS) * time.Millisecond)
defer ticker.Stop()
for {
select {
case <-c.stopChan:
return
case <-ticker.C:
c.runCleanup()
}
}
}
// runCleanup scans the services directory and removes dead services
func (c *Cleaner) runCleanup() {
c.mu.Lock()
c.stats.TotalRuns++
c.stats.LastRunTime = time.Now().UTC().Format(time.RFC3339)
c.mu.Unlock()
entries, err := os.ReadDir(c.servicesDir)
if err != nil {
if !os.IsNotExist(err) {
log.Printf("[Cleaner] Failed to read services directory: %v", err)
c.mu.Lock()
c.stats.Errors++
c.mu.Unlock()
}
return
}
threshold := time.Duration(c.config.DeadServiceThresholdMS) * time.Millisecond
removedCount := 0
for _, entry := range entries {
if entry.IsDir() || filepath.Ext(entry.Name()) != ".json" {
continue
}
filePath := filepath.Join(c.servicesDir, entry.Name())
// Never delete our own service file
if filePath == c.ownFile {
continue
}
// Check if service is dead
isDead, serviceName, age := c.isServiceDead(filePath, threshold)
if isDead {
if err := os.Remove(filePath); err != nil {
log.Printf("[Cleaner] Failed to remove dead service file %s: %v", entry.Name(), err)
c.mu.Lock()
c.stats.Errors++
c.mu.Unlock()
} else {
ageMinutes := age.Minutes()
log.Printf("[Cleaner] Removed dead service: %s (last heartbeat: %.1f minutes ago)", serviceName, ageMinutes)
removedCount++
}
}
}
if removedCount > 0 {
c.mu.Lock()
c.stats.ServicesRemoved += removedCount
c.mu.Unlock()
log.Printf("[Cleaner] Cleanup complete: removed %d dead service(s)", removedCount)
}
}
// isServiceDead checks if a service file represents a dead service
// Returns: isDead, serviceName, age since last heartbeat
func (c *Cleaner) isServiceDead(filePath string, threshold time.Duration) (bool, string, time.Duration) {
data, err := os.ReadFile(filePath)
if err != nil {
// Can't read file - treat as dead to clean up corrupted files
log.Printf("[Cleaner] Cannot read service file %s: %v (treating as dead)", filepath.Base(filePath), err)
return true, filepath.Base(filePath), threshold
}
var info ServiceInfo
if err := json.Unmarshal(data, &info); err != nil {
// Corrupted JSON - skip rather than delete (might be recoverable)
log.Printf("[Cleaner] Skipping corrupted service file %s: %v", filepath.Base(filePath), err)
return false, "", 0
}
// Parse last heartbeat timestamp
if info.LastHeartbeat == "" {
// No heartbeat timestamp - treat as dead
return true, info.Name, threshold
}
lastHeartbeat, err := time.Parse(time.RFC3339, info.LastHeartbeat)
if err != nil {
// Invalid timestamp - treat as dead
log.Printf("[Cleaner] Invalid timestamp in service file %s: %v (treating as dead)", filepath.Base(filePath), err)
return true, info.Name, threshold
}
age := time.Since(lastHeartbeat)
if age > threshold {
return true, info.Name, age
}
return false, info.Name, age
}
// Stats returns a copy of the current cleanup statistics
func (c *Cleaner) Stats() CleanupStats {
c.mu.RLock()
defer c.mu.RUnlock()
return CleanupStats{
LastRunTime: c.stats.LastRunTime,
ServicesRemoved: c.stats.ServicesRemoved,
TotalRuns: c.stats.TotalRuns,
Errors: c.stats.Errors,
}
}
-286
View File
@@ -1,286 +0,0 @@
package discovery
import (
"encoding/json"
"fmt"
"log"
"net"
"os"
"path/filepath"
"sync"
"time"
"github.com/metazla/meta-core/internal/config"
)
// ServiceInfo contains simplified service registration data
// URLs are obtained via the /urls API endpoint
type ServiceInfo struct {
Name string `json:"name"`
Hostname string `json:"hostname"`
BaseUrl string `json:"baseUrl"`
Status string `json:"status"`
LastHeartbeat string `json:"lastHeartbeat"`
}
// Service handles service registration and discovery
type Service struct {
config *config.Config
servicesDir string
serviceFile string
info *ServiceInfo
mu sync.RWMutex
stopChan chan struct{}
wg sync.WaitGroup
}
// NewService creates a new service discovery instance
func NewService(cfg *config.Config) *Service {
hostname, _ := os.Hostname()
return &Service{
config: cfg,
servicesDir: cfg.ServicesDir(),
serviceFile: filepath.Join(cfg.ServicesDir(), cfg.ServiceName+"-"+hostname+".json"),
stopChan: make(chan struct{}),
}
}
// Start begins service registration and heartbeat
func (s *Service) Start() error {
log.Printf("[Discovery] Starting service discovery for %s", s.config.ServiceName)
// Ensure services directory exists
if err := os.MkdirAll(s.servicesDir, 0755); err != nil {
return fmt.Errorf("failed to create services directory: %w", err)
}
// Build and register service info
s.info = s.buildServiceInfo()
if err := s.register(); err != nil {
return fmt.Errorf("failed to register service: %w", err)
}
// Start heartbeat loop
s.wg.Add(1)
go s.heartbeatLoop()
return nil
}
// Stop stops service discovery and unregisters
func (s *Service) Stop() error {
log.Println("[Discovery] Stopping service discovery...")
close(s.stopChan)
s.wg.Wait()
// Unregister by removing service file
if err := os.Remove(s.serviceFile); err != nil && !os.IsNotExist(err) {
log.Printf("[Discovery] Failed to remove service file: %v", err)
}
return nil
}
// buildServiceInfo creates the service info for this instance
func (s *Service) buildServiceInfo() *ServiceInfo {
hostname, _ := os.Hostname()
ip := getLocalIP()
// Browser-facing nav URL: prefer META_CORE_PUBLIC_URL (the reachable URL
// when there's no Caddy perimeter in front — e.g. a debug-direct port),
// then BASE_URL (the Caddy perimeter URL), then the internal IP.
var baseUrl string
if s.config.PublicURL != "" {
baseUrl = s.config.PublicURL
} else if s.config.BaseURL != "" {
baseUrl = s.config.BaseURL
} else {
baseUrl = fmt.Sprintf("http://%s:%d", ip, s.config.APIPort)
}
info := &ServiceInfo{
Name: s.config.ServiceName,
Hostname: hostname,
BaseUrl: baseUrl,
Status: "running",
LastHeartbeat: time.Now().UTC().Format(time.RFC3339),
}
return info
}
// register writes service info to file
func (s *Service) register() error {
s.mu.Lock()
defer s.mu.Unlock()
return s.writeServiceInfo(s.info)
}
// writeServiceInfo atomically writes service info to file
func (s *Service) writeServiceInfo(info *ServiceInfo) error {
tempPath := s.serviceFile + ".tmp"
data, err := json.MarshalIndent(info, "", " ")
if err != nil {
return err
}
if err := os.WriteFile(tempPath, data, 0644); err != nil {
return err
}
return os.Rename(tempPath, s.serviceFile)
}
// heartbeatLoop periodically updates the last heartbeat
func (s *Service) heartbeatLoop() {
defer s.wg.Done()
ticker := time.NewTicker(time.Duration(s.config.HeartbeatIntervalMS) * time.Millisecond)
defer ticker.Stop()
for {
select {
case <-s.stopChan:
return
case <-ticker.C:
s.heartbeat()
}
}
}
// heartbeat updates the last heartbeat timestamp
func (s *Service) heartbeat() {
s.mu.Lock()
defer s.mu.Unlock()
if s.info == nil {
return
}
s.info.LastHeartbeat = time.Now().UTC().Format(time.RFC3339)
if err := s.writeServiceInfo(s.info); err != nil {
log.Printf("[Discovery] Failed to update heartbeat: %v", err)
}
}
// UpdateStatus updates the service status
func (s *Service) UpdateStatus(status string) error {
s.mu.Lock()
defer s.mu.Unlock()
if s.info == nil {
return fmt.Errorf("service not registered")
}
s.info.Status = status
s.info.LastHeartbeat = time.Now().UTC().Format(time.RFC3339)
return s.writeServiceInfo(s.info)
}
// Discover finds a service by name
// Looks for files matching pattern: name-*.json (hostname-based naming)
func (s *Service) Discover(name string) (*ServiceInfo, error) {
// First try exact match for backward compatibility
exactPath := filepath.Join(s.servicesDir, name+".json")
if data, err := os.ReadFile(exactPath); err == nil {
var info ServiceInfo
if err := json.Unmarshal(data, &info); err == nil {
return s.checkStale(&info), nil
}
}
// Search for hostname-based files: name-*.json
pattern := filepath.Join(s.servicesDir, name+"-*.json")
matches, err := filepath.Glob(pattern)
if err != nil {
return nil, err
}
// Return the first valid service found
for _, match := range matches {
data, err := os.ReadFile(match)
if err != nil {
continue
}
var info ServiceInfo
if err := json.Unmarshal(data, &info); err != nil {
continue
}
return s.checkStale(&info), nil
}
return nil, nil
}
// checkStale marks a service as stale if heartbeat is too old
func (s *Service) checkStale(info *ServiceInfo) *ServiceInfo {
lastHeartbeat, err := time.Parse(time.RFC3339, info.LastHeartbeat)
if err == nil {
staleThreshold := time.Duration(s.config.StaleThresholdMS) * time.Millisecond
if time.Since(lastHeartbeat) > staleThreshold {
info.Status = "stale"
}
}
return info
}
// DiscoverAll finds all registered services
func (s *Service) DiscoverAll() ([]*ServiceInfo, error) {
entries, err := os.ReadDir(s.servicesDir)
if err != nil {
if os.IsNotExist(err) {
return []*ServiceInfo{}, nil
}
return nil, err
}
var services []*ServiceInfo
for _, entry := range entries {
if entry.IsDir() || filepath.Ext(entry.Name()) != ".json" {
continue
}
filePath := filepath.Join(s.servicesDir, entry.Name())
data, err := os.ReadFile(filePath)
if err != nil {
log.Printf("[Discovery] Failed to read service file %s: %v", entry.Name(), err)
continue
}
var info ServiceInfo
if err := json.Unmarshal(data, &info); err != nil {
log.Printf("[Discovery] Failed to parse service file %s: %v", entry.Name(), err)
continue
}
services = append(services, s.checkStale(&info))
}
return services, nil
}
// getLocalIP returns the local IP address
func getLocalIP() string {
addrs, err := net.InterfaceAddrs()
if err != nil {
hostname, _ := os.Hostname()
return hostname
}
for _, addr := range addrs {
if ipnet, ok := addr.(*net.IPNet); ok && !ipnet.IP.IsLoopback() {
if ipnet.IP.To4() != nil {
return ipnet.IP.String()
}
}
}
hostname, _ := os.Hostname()
return hostname
}
+434
View File
@@ -0,0 +1,434 @@
package meshdisco
import (
"context"
"encoding/json"
"fmt"
"log"
"net"
"os"
"sort"
"sync"
"syscall"
"time"
"golang.org/x/net/ipv4"
)
// Config configures a Node. Payload is a function rather than a value so the
// announce is rebuilt from live state on every tick — that is what keeps the
// announced URLs and GET /urls from drifting apart.
type Config struct {
Name string
Instance string
Role string
Version string
Group string
Port int
Interval time.Duration
// Payload returns the identity/URL fields for this announce. Name,
// Instance, Role, Version and the envelope are filled in by the Node.
Payload func() (baseURL string, status string, urls *URLs)
}
func (c *Config) applyDefaults() {
if c.Group == "" {
c.Group = DefaultGroup
}
if c.Port == 0 {
c.Port = DefaultPort
}
if c.Interval <= 0 {
c.Interval = DefaultInterval
}
if c.Role == "" {
c.Role = RoleService
}
}
// Node is one participant in the mesh: it announces itself, answers probes,
// and keeps a TTL map of everyone else it has heard from.
type Node struct {
cfg Config
group net.IP
pc *ipv4.PacketConn
closer interface{ Close() error }
// ifaces are the multicast-capable interfaces we joined and send on.
// A container attached to two docker networks (metashare-app on `pcs`
// and `metamesh-mesh`, metawatch-* on `metawatch` and `metamesh-mesh`)
// has two entries here, and MUST be announced on both — sending via the
// default route alone silently covers only one network.
ifaces []net.Interface
// sendMu serialises SetMulticastInterface + WriteTo, which are a
// two-step operation on one shared socket.
sendMu sync.Mutex
mu sync.RWMutex
neighbors map[string]*Neighbor
stopChan chan struct{}
stopOnce sync.Once
wg sync.WaitGroup
}
// New creates a Node. It does not touch the network until Start.
func New(cfg Config) *Node {
cfg.applyDefaults()
return &Node{
cfg: cfg,
neighbors: make(map[string]*Neighbor),
stopChan: make(chan struct{}),
}
}
// Start opens the multicast socket, joins the group on every eligible
// interface, and begins announcing. A join failure on one interface is logged
// and skipped rather than fatal — matching the ENABLE_MDNS precedent in the
// Rust services, where a host that blocks multicast degrades instead of
// failing to boot.
func (n *Node) Start() error {
n.group = net.ParseIP(n.cfg.Group)
if n.group == nil || n.group.To4() == nil {
return fmt.Errorf("meshdisco: invalid multicast group %q", n.cfg.Group)
}
lc := net.ListenConfig{Control: reusePort}
conn, err := lc.ListenPacket(context.Background(), "udp4", fmt.Sprintf(":%d", n.cfg.Port))
if err != nil {
return fmt.Errorf("meshdisco: listen on :%d: %w", n.cfg.Port, err)
}
n.closer = conn
pc := ipv4.NewPacketConn(conn)
// TTL 1: link-local only. Discovery never routes off the local segment;
// the two-host topology keeps two separate meshes by design.
if err := pc.SetMulticastTTL(1); err != nil {
log.Printf("[meshdisco] Warning: could not set multicast TTL: %v", err)
}
n.pc = pc
n.ifaces = eligibleInterfaces()
if len(n.ifaces) == 0 {
log.Println("[meshdisco] Warning: no multicast-capable interfaces found; discovery will be inert")
}
group := &net.UDPAddr{IP: n.group}
joined := 0
for i := range n.ifaces {
if err := pc.JoinGroup(&n.ifaces[i], group); err != nil {
log.Printf("[meshdisco] Warning: join %s on %s failed: %v", n.cfg.Group, n.ifaces[i].Name, err)
continue
}
joined++
}
log.Printf("[meshdisco] Listening on %s:%d as %s/%s (joined %d of %d interfaces)",
n.cfg.Group, n.cfg.Port, n.cfg.Name, n.cfg.Instance, joined, len(n.ifaces))
n.wg.Add(2)
go n.readLoop()
go n.announceLoop()
return nil
}
// Stop closes the socket and waits for the loops to exit. It sends a final
// announce with status "stopping" so neighbours drop us immediately instead of
// waiting out the staleness window.
func (n *Node) Stop() error {
var err error
n.stopOnce.Do(func() {
n.sendAnnounce(nil, "stopping")
close(n.stopChan)
if n.closer != nil {
err = n.closer.Close()
}
n.wg.Wait()
})
return err
}
// Probe multicasts a discovery message on every interface. Every listener
// replies immediately, so a caller does not have to wait out an announce
// interval. This is what makes the protocol usable on the boot path.
func (n *Node) Probe() {
n.broadcast(Message{V: ProtocolVersion, Type: TypeDiscovery})
}
// Neighbors returns everyone heard from within the staleness window, sorted by
// name. Entries older than that are dropped as a side effect — expiry is a
// read-time concern, so there is no reaper goroutine and no equivalent of the
// old cleaner.go.
func (n *Node) Neighbors() []Neighbor {
cutoff := time.Now().Add(-n.cfg.Interval * LivenessFactor)
n.mu.Lock()
out := make([]Neighbor, 0, len(n.neighbors))
for k, nb := range n.neighbors {
if nb.LastSeen.Before(cutoff) {
delete(n.neighbors, k)
continue
}
out = append(out, *nb)
}
n.mu.Unlock()
sort.Slice(out, func(i, j int) bool {
if out[i].Name != out[j].Name {
return out[i].Name < out[j].Name
}
return out[i].Instance < out[j].Instance
})
return out
}
// NeighborsByName collapses multiple instances of one service down to the most
// recently seen, which is the shape the nav menu wants (and what the existing
// /api/services dedup contract asserts).
func (n *Node) NeighborsByName() []Neighbor {
best := make(map[string]Neighbor)
for _, nb := range n.Neighbors() {
if cur, ok := best[nb.Name]; !ok || nb.LastSeen.After(cur.LastSeen) {
best[nb.Name] = nb
}
}
out := make([]Neighbor, 0, len(best))
for _, nb := range best {
out = append(out, nb)
}
sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })
return out
}
// Self returns this node's own announce, so callers can render themselves in
// the menu without waiting to hear their own multicast echo back.
func (n *Node) Self() Neighbor {
return Neighbor{Message: n.buildAnnounce("running"), LastSeen: time.Now()}
}
func (n *Node) readLoop() {
defer n.wg.Done()
buf := make([]byte, ReadBufferSize)
for {
select {
case <-n.stopChan:
return
default:
}
nRead, _, src, err := n.pc.ReadFrom(buf)
if err != nil {
select {
case <-n.stopChan:
return
default:
}
log.Printf("[meshdisco] Read error: %v", err)
continue
}
var msg Message
if err := json.Unmarshal(buf[:nRead], &msg); err != nil {
// Malformed or foreign traffic on the group — ignore quietly.
continue
}
if msg.V != ProtocolVersion {
continue
}
// Drop our own multicast echo. Loopback is left enabled so that a
// single-container test still sees traffic; we filter by identity
// instead of relying on the socket option.
if msg.Instance != "" && msg.Instance == n.cfg.Instance && msg.Name == n.cfg.Name {
continue
}
switch msg.Type {
case TypeDiscovery:
// Unicast the reply straight back to the prober's source
// address. The kernel picks the right interface by route, and
// the prober's ephemeral source port is where it is listening.
n.sendAnnounce(src, "running")
case TypeAnnounce:
n.record(msg, src)
}
}
}
func (n *Node) record(msg Message, src net.Addr) {
addr := ""
if src != nil {
if udp, ok := src.(*net.UDPAddr); ok {
addr = udp.IP.String()
} else {
addr = src.String()
}
}
if msg.Status == "stopping" {
n.mu.Lock()
delete(n.neighbors, msg.key())
n.mu.Unlock()
return
}
nb := &Neighbor{Message: msg, Addr: addr, LastSeen: time.Now()}
n.mu.Lock()
n.neighbors[msg.key()] = nb
n.mu.Unlock()
}
func (n *Node) announceLoop() {
defer n.wg.Done()
// Announce immediately, then probe once so the map fills without waiting
// a full interval for everyone else's ticker.
n.sendAnnounce(nil, "running")
n.Probe()
ticker := time.NewTicker(n.cfg.Interval)
defer ticker.Stop()
for {
select {
case <-n.stopChan:
return
case <-ticker.C:
n.sendAnnounce(nil, "running")
}
}
}
func (n *Node) buildAnnounce(status string) Message {
msg := Message{
V: ProtocolVersion,
Type: TypeAnnounce,
Name: n.cfg.Name,
Instance: n.cfg.Instance,
Role: n.cfg.Role,
Version: n.cfg.Version,
Status: status,
}
if n.cfg.Payload != nil {
baseURL, st, urls := n.cfg.Payload()
msg.BaseUrl = baseURL
if st != "" && status == "running" {
msg.Status = st
}
// Only meta-core carries a URLs block; a service announcing one
// would let any container on the network impersonate the core.
if n.cfg.Role == RoleCore {
msg.URLs = urls
}
}
return msg
}
// sendAnnounce multicasts when dst is nil, or unicasts to dst (a probe reply).
func (n *Node) sendAnnounce(dst net.Addr, status string) {
msg := n.buildAnnounce(status)
if dst == nil {
n.broadcast(msg)
return
}
body, err := json.Marshal(msg)
if err != nil {
return
}
n.sendMu.Lock()
defer n.sendMu.Unlock()
if _, err := n.pc.WriteTo(body, nil, dst); err != nil {
log.Printf("[meshdisco] Reply to %s failed: %v", dst, err)
}
}
// broadcast writes one message to the group once per eligible interface.
// Writing once and letting the routing table choose would reach exactly one
// network, which is the failure mode this protocol exists to avoid.
func (n *Node) broadcast(msg Message) {
if n.pc == nil {
return
}
body, err := json.Marshal(msg)
if err != nil {
return
}
dst := &net.UDPAddr{IP: n.group, Port: n.cfg.Port}
n.sendMu.Lock()
defer n.sendMu.Unlock()
for i := range n.ifaces {
if err := n.pc.SetMulticastInterface(&n.ifaces[i]); err != nil {
log.Printf("[meshdisco] Warning: select interface %s failed: %v", n.ifaces[i].Name, err)
continue
}
if _, err := n.pc.WriteTo(body, nil, dst); err != nil {
log.Printf("[meshdisco] Warning: send on %s failed: %v", n.ifaces[i].Name, err)
}
}
}
// eligibleInterfaces returns every up, non-loopback, multicast-capable
// interface that has an IPv4 address.
func eligibleInterfaces() []net.Interface {
all, err := net.Interfaces()
if err != nil {
log.Printf("[meshdisco] Warning: could not enumerate interfaces: %v", err)
return nil
}
var out []net.Interface
for _, ifi := range all {
if ifi.Flags&net.FlagUp == 0 ||
ifi.Flags&net.FlagLoopback != 0 ||
ifi.Flags&net.FlagMulticast == 0 {
continue
}
addrs, err := ifi.Addrs()
if err != nil {
continue
}
for _, a := range addrs {
if ipnet, ok := a.(*net.IPNet); ok && ipnet.IP.To4() != nil {
out = append(out, ifi)
break
}
}
}
return out
}
// reusePort sets SO_REUSEADDR so a second listener (a test, or a sidecar) can
// bind the same multicast port on the same host.
func reusePort(_, _ string, c syscall.RawConn) error {
var serr error
err := c.Control(func(fd uintptr) {
serr = syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, syscall.SO_REUSEADDR, 1)
})
if err != nil {
return err
}
return serr
}
// LocalIPv4 returns this host's first non-loopback IPv4 address, or the
// hostname when none can be determined. Same fallback the file-based registry
// used (discovery/service.go), kept identical so announced URLs don't shift.
func LocalIPv4() string {
addrs, err := net.InterfaceAddrs()
if err == nil {
for _, addr := range addrs {
if ipnet, ok := addr.(*net.IPNet); ok && !ipnet.IP.IsLoopback() {
if ipnet.IP.To4() != nil {
return ipnet.IP.String()
}
}
}
}
host, _ := os.Hostname()
return host
}
+132
View File
@@ -0,0 +1,132 @@
// Package meshdisco implements meta-discovery v1: UDP multicast service
// discovery for the meta-* services.
//
// It replaces two file-based mechanisms that both required a shared
// /meta-core volume:
//
// - /meta-core/locks/kv-leader.info — a one-line file holding meta-core's
// API URL, read at boot to locate meta-core.
// - /meta-core/services/<name>-<host>.json — per-service registration files
// aggregated into the dashboard's service nav.
//
// Both answer the same question — where is meta-core, and who else is on this
// network — so one announce packet answers both. Discovery scope becomes the
// docker network the container is attached to rather than the volume it
// mounts, which is what lets per-app CasaOS stacks and the meta-watch island
// see their neighbours without sharing storage.
//
// The wire protocol is documented in
// docs/project-architecture/service-discovery.md. Keep this file and that doc
// in step — every other language implementation is written against the doc.
package meshdisco
import "time"
const (
// ProtocolVersion is the `v` field on every message. Receivers ignore
// messages carrying a version they don't know.
ProtocolVersion = 1
// DefaultGroup is the IPv4 multicast group. Deliberately NOT beacon's
// 239.255.99.1 (sandbox/beacon), so the two protocols cannot cross-talk
// if they ever share a network.
DefaultGroup = "239.255.77.1"
// DefaultPort is the UDP port. Deliberately NOT beacon's 9099, for the
// same reason.
DefaultPort = 9399
// DefaultInterval is how often an unsolicited announce is emitted.
DefaultInterval = 10 * time.Second
// LivenessFactor multiplies the announce interval to get the staleness
// window: a neighbour last seen longer ago than interval*factor is
// dropped. This replaces the old heartbeat file + cleaner.go reaper.
LivenessFactor = 3
// ReadBufferSize is the datagram receive buffer. Beacon's reference
// implementation uses 1 KiB, which truncates a realistic payload; an
// announce carrying the full URLs block is comfortably under 8 KiB.
ReadBufferSize = 8192
// TypeDiscovery is a probe: "who is out there?". Sent at startup and on
// a manual refresh, answered immediately by every listener.
TypeDiscovery = "discovery"
// TypeAnnounce is a service describing itself, sent unsolicited on a
// ticker and unicast in reply to a probe.
TypeAnnounce = "announce"
// RoleCore marks meta-core itself — the only role that carries a URLs
// block. Consumers locating meta-core filter on this.
RoleCore = "core"
// RoleService marks every other meta-* service.
RoleService = "service"
)
// URLs mirrors meta-core's existing GET /urls response (api.URLsResponse)
// minus redisUrl, which the api-mediated-access lockdown retired. Embedding it
// in the announce makes the announce a drop-in for the /urls hop that
// LeaderClient performs today; /urls itself stays, since meta-search and
// meta-share still call it.
type URLs struct {
Hostname string `json:"hostname"`
BaseUrl string `json:"baseUrl"`
ApiUrl string `json:"apiUrl"`
WebdavUrl string `json:"webdavUrl"`
WebdavUrlInternal string `json:"webdavUrlInternal"`
}
// Message is the single wire type: both probe and announce are one JSON
// datagram. Everything but `v` and `type` is omitted on a probe.
type Message struct {
V int `json:"v"`
Type string `json:"type"`
// Name is the service name, e.g. "meta-core", "meta-sort".
Name string `json:"name,omitempty"`
// Instance distinguishes two instances of the same Name — the container
// hostname. Also used to recognise and drop our own multicast echo.
Instance string `json:"instance,omitempty"`
// Role is RoleCore or RoleService.
Role string `json:"role,omitempty"`
// Version is the service's build version, for display only.
Version string `json:"version,omitempty"`
// Status is "running" | "starting" | "stopping".
Status string `json:"status,omitempty"`
// BaseUrl is the browser-facing URL the nav menu links to. Built with
// the same three-way fallback the file registry used:
// META_CORE_PUBLIC_URL -> BASE_URL -> http://<local-ip>:<port>.
BaseUrl string `json:"baseUrl,omitempty"`
// URLs is populated only when Role == RoleCore.
URLs *URLs `json:"urls,omitempty"`
// Token is reserved for a future HMAC over the payload. v1 senders omit
// it and v1 receivers ignore it; see the doc's Security section.
Token string `json:"token,omitempty"`
}
// Neighbor is a Message plus the receive-side facts the sender cannot state
// about itself.
type Neighbor struct {
Message
// Addr is the source address the datagram arrived from. Taken from the
// packet, never from the payload — a sender does not get to claim its
// own IP.
Addr string `json:"addr"`
// LastSeen is when we last received an announce from this instance.
LastSeen time.Time `json:"lastSeen"`
}
// key identifies a neighbour entry. Two instances of the same service are
// distinct neighbours; callers that want one row per service (the nav menu)
// collapse by Name themselves.
func (m Message) key() string {
if m.Instance == "" {
return m.Name
}
return m.Name + "|" + m.Instance
}
+47
View File
@@ -422,3 +422,50 @@ func (c *Client) GetMetadataDocument(uuid string) (*MetadataDocument, error) {
Duplicates: dups,
}, nil
}
// RegisterFileTuple records one physical file under its midhash256 CID, the
// same way the file watcher does for every file it scans: mint a UUID root
// (filePath/sizeByte/mtimeNano) and write `cids/<midhash>`, whose hook
// registers the cid:<midhash> reverse alias. Content already known under
// another path is recorded as a duplicate; the same path just refreshes
// size/mtime.
//
// Returns the root and whether it was newly minted. Shared by the watcher and
// POST /api/files/register, which is how a box running with the watcher off
// (a meta-watch client core) still gets files it writes under /files/plugin
// resolvable by CID.
func (c *Client) RegisterFileTuple(absPath string, size, mtimeNano int64, midhash string) (string, bool, error) {
if midhash == "" {
return "", false, fmt.Errorf("midhash is required")
}
uuid, err := c.GetByCID(midhash)
if err != nil {
return "", false, fmt.Errorf("GetByCID(%s): %w", midhash, err)
}
if uuid == "" {
uuid, err = c.Mint(absPath, size, mtimeNano)
if err != nil {
return "", false, fmt.Errorf("Mint(%s): %w", absPath, err)
}
if err := c.SetProperty(uuid, CIDsKeyPrefix+midhash, "true"); err != nil {
return uuid, true, fmt.Errorf("SetProperty cids/%s for %s: %w", midhash, uuid, err)
}
return uuid, true, nil
}
existing, _ := c.GetProperty(uuid, "filePath")
if existing != absPath {
if _, err := c.AddDuplicatePath(uuid, absPath); err != nil {
return uuid, false, fmt.Errorf("AddDuplicatePath %s @ %s: %w", uuid, absPath, err)
}
return uuid, false, nil
}
if _, err := c.MergeMetadataFlat(uuid, map[string]string{
"sizeByte": strconv.FormatInt(size, 10),
"mtimeNano": strconv.FormatInt(mtimeNano, 10),
}); err != nil {
return uuid, false, fmt.Errorf("refresh tuple for %s: %w", uuid, err)
}
return uuid, false, nil
}
+4 -4
View File
@@ -59,7 +59,7 @@ func TestFieldIndex_SetMetadataFlatIndexesFields(t *testing.T) {
"filePath": "/files/watch/a.mkv",
"sizeByte": "123",
"cids/bafkreiaaa": "true",
"titles/jpn": "ナルト",
"titles/jpn/ナルト": "true",
"categories/TV/UHD": "true",
"info/files/0/path": "a.mkv",
"provenance": `{"source":"gateway"}`,
@@ -380,7 +380,7 @@ func TestFieldIndex_WarmBuildsEveryMissingIndex(t *testing.T) {
// written through the normal path (already indexed).
legacy := map[string]map[string]string{
"old1": {"filePath": "/a.mkv", "cids/bafkreiaaa": "true"},
"old2": {"filePath": "/b.mkv", "titles/jpn": "ナルト", "deep/a/b": "x"},
"old2": {"filePath": "/b.mkv", "titles/jpn/ナルト": "true", "deep/a/b": "x"},
}
for root, fields := range legacy {
for f, v := range fields {
@@ -407,7 +407,7 @@ func TestFieldIndex_WarmBuildsEveryMissingIndex(t *testing.T) {
if got := fieldNames(t, c, "old1"); !reflect.DeepEqual(got, []string{"cids/bafkreiaaa", "filePath"}) {
t.Errorf("old1 index = %v", got)
}
if got := fieldNames(t, c, "old2"); !reflect.DeepEqual(got, []string{"deep/a/b", "filePath", "titles/jpn"}) {
if got := fieldNames(t, c, "old2"); !reflect.DeepEqual(got, []string{"deep/a/b", "filePath", "titles/jpn/ナルト"}) {
t.Errorf("old2 index = %v", got)
}
if got := fieldNames(t, c, "fresh"); !reflect.DeepEqual(got, []string{"title"}) {
@@ -425,7 +425,7 @@ func TestFieldIndex_WarmBuildsEveryMissingIndex(t *testing.T) {
if _, err := c.WarmFieldIndexes(); err != nil {
t.Fatalf("second WarmFieldIndexes: %v", err)
}
if got := fieldNames(t, c, "old2"); !reflect.DeepEqual(got, []string{"deep/a/b", "filePath", "titles/jpn"}) {
if got := fieldNames(t, c, "old2"); !reflect.DeepEqual(got, []string{"deep/a/b", "filePath", "titles/jpn/ナルト"}) {
t.Errorf("old2 index after second warm = %v", got)
}
})
+326
View File
@@ -0,0 +1,326 @@
package storage
import (
"context"
"fmt"
"strings"
"time"
"github.com/redis/go-redis/v9"
"github.com/metazla/meta-core/internal/cid"
)
// The one-shot corpus sweep for meta-read `docs/reading-model.md` §7:
// identity and position keys are resolved, never copied from a query, and a
// card carries no position keys and no `workcid`.
//
// # Why a sweep
//
// A feeder answering a `workcid:` / `anilistid:` query could stamp the query's
// own filter values onto what it returned. The gateway persists whatever passes
// the filter, so the copy became a permanent binding — and stored records never
// pick up the writer fix: a covered `(upstream, query)` never reaches the
// feeder again, and the streaming dedup keeps meta-core's copy over a fresh
// one. So the copies already at rest are removed here, once.
//
// What the copies cost: an `anilistid` on a MangaDex site card claims an
// AniList binding nobody resolved, and meta-read reads a bare `anilistid` as an
// anchor — the site card then competes with (and can hide) the real AniList
// card for the same work. A `workcid` or `chapterNumber` on a card makes a work
// join and sort like a position inside one.
//
// # What is swept, and what deliberately is not
//
// - Only `domain=literature` + `fileType=card` records. Nothing else is read
// for writing, whatever keys it carries.
// - `workcid`, `chapterNumber`, `volumeNumber`, `chapterStart`, `chapterEnd`
// go from **every** such card: a card is a work, never a position in one,
// so no legitimate writer puts them there.
// - `anilistid` goes only from a **site card** — a card whose own `cids/`
// key-set holds a site-form card locator and no locator of any other form.
// On an anchor card the id *is* the card's identity (its locator is
// `("anilist", …)`). A card holding both forms (a cross-source merge) keeps
// it: which half the id belongs to is not recoverable at rest, and leaving
// a stamp is the safe direction.
// - **Chapters are never swept.** A bare `anilistid` on a chapter is graded
// Matched by rule, and a query-copied stamp is indistinguishable at rest
// from a title-matched one — deleting would drop real bindings with the
// fake ones.
//
// # Contract
//
// Dry-run unless `apply`: the sweep deletes keys, so the default call only
// reports. Idempotent: a swept record plans nothing on the next pass, so the
// endpoint is safe to call repeatedly and on a fresh box.
// literatureEchoAnchorSources are the card-locator sources that name an id-bag
// work (an anchor), never a site.
//
// ⚠ Mirror of meta-read `card_cid::is_host_form`. meta-read spells out only
// `anilist` and `openlibrary`; the others carry no '.', so the '.' test already
// excludes them and the two rules agree on every locator minted today. They are
// listed so the rule reads on its own here.
var literatureEchoAnchorSources = map[string]bool{
"anilist": true,
"openlibrary": true,
"tmdb": true,
"imdb": true,
"musicbrainz": true,
"mal": true,
}
// literatureCardEchoFields are removed from every literature card carrying
// them — position keys and the work reference, none of which a card may hold.
var literatureCardEchoFields = []string{
"workcid",
"chapterNumber",
"volumeNumber",
"chapterStart",
"chapterEnd",
}
// literatureEchoReadFields is the per-record MGET layout: the two gate fields,
// then `anilistid`, then the echo fields in literatureCardEchoFields order.
var literatureEchoReadFields = append(
[]string{"domain", "fileType", "anilistid"},
literatureCardEchoFields...,
)
// LiteratureEchoSweep is the result of one sweep pass. Planned and Fixed count
// records; ByKey counts fields per key — removed on an apply pass, that would
// be removed on a dry run. ByKey is never nil so it serialises as `{}`.
type LiteratureEchoSweep struct {
Planned int
Fixed int
ByKey map[string]int
}
// literatureEchoPlan is one record's pending deletes, in a fixed order
// (`anilistid` first, then literatureCardEchoFields order).
type literatureEchoPlan struct {
hashID string
fields []string
}
// isSiteFormCardSource is the site-form rule on an already-decoded source: not
// an anchor source, and shaped like a host.
func isSiteFormCardSource(source string) bool {
return !literatureEchoAnchorSources[source] && strings.Contains(source, ".")
}
// isSiteFormCardLocator reports whether a bare CID is a card locator (0x1007)
// whose source is a site (`mangadex.org`), not an anchor (`anilist`).
func isSiteFormCardLocator(cidStr string) bool {
source, _, err := cid.DecodeCard(cidStr)
return err == nil && isSiteFormCardSource(source)
}
// isSiteCard reports whether a card's field names prove it a site card: at
// least one site-form `cids/` card locator and no card locator of another form.
// Non-card members (a content CID) say nothing about which work the card is and
// are ignored. An empty field list — a record with no field-name index — is
// never a site card, which keeps its `anilistid`; see planLiteratureEchoes.
func isSiteCard(fields []string) bool {
site := false
for _, f := range fields {
member := cidFromKeysetField(f)
if member == "" {
continue
}
source, _, err := cid.DecodeCard(member)
if err != nil {
continue
}
if !isSiteFormCardSource(source) {
return false
}
site = true
}
return site
}
// SweepLiteratureEchoes removes query echoes stored on literature cards (see
// the file comment). With apply=false it only plans.
//
// Reads in batched passes (never a per-record SCAN — a single record read
// costs a full keyspace sweep on this store), then deletes only the planned
// fields, through DeleteProperty so the field-name index and the search-dirty
// set stay consistent.
func (c *Client) SweepLiteratureEchoes(ctx context.Context, apply bool) (LiteratureEchoSweep, error) {
res := LiteratureEchoSweep{ByKey: map[string]int{}}
plans, err := c.planLiteratureEchoes(ctx)
if err != nil {
return res, err
}
res.Planned = len(plans)
if !apply {
for _, p := range plans {
for _, f := range p.fields {
res.ByKey[f]++
}
}
return res, nil
}
// Writes take the client lock themselves, so they happen after the read
// phase has released it. ByKey counts what was actually deleted, so a
// failure midway reports the true partial state.
for _, p := range plans {
// DeleteProperty carries its own per-call timeout; checking ctx here is
// what makes the wrapper's deadline bound the whole write phase.
if err := ctx.Err(); err != nil {
return res, err
}
for _, f := range p.fields {
if err := c.DeleteProperty(p.hashID, f); err != nil {
return res, fmt.Errorf("delete %s on %s: %w", f, p.hashID, err)
}
res.ByKey[f]++
}
res.Fixed++
}
return res, nil
}
// planLiteratureEchoes is the read half: what would be deleted, without
// deleting it.
func (c *Client) planLiteratureEchoes(ctx context.Context) ([]literatureEchoPlan, error) {
c.mu.RLock()
defer c.mu.RUnlock()
if c.client == nil {
return nil, fmt.Errorf("not connected")
}
hashIDs, err := c.client.SMembers(ctx, c.buildIndexKey()).Result()
if err != nil {
return nil, fmt.Errorf("smembers index: %w", err)
}
if len(hashIDs) == 0 {
return nil, nil
}
// Pass 1: the gate and echo fields for every record, in a fixed order so
// the MGET response decodes by position: keys[i*per+n] belongs to
// hashIDs[i].
per := len(literatureEchoReadFields)
keys := make([]string, 0, len(hashIDs)*per)
for _, h := range hashIDs {
prefix := c.buildKeyPrefix(h)
for _, f := range literatureEchoReadFields {
keys = append(keys, prefix+f)
}
}
// Chunked so one MGET never carries the whole corpus.
const chunk = 3000
values := make([]interface{}, 0, len(keys))
for start := 0; start < len(keys); start += chunk {
end := start + chunk
if end > len(keys) {
end = len(keys)
}
got, err := c.client.MGet(ctx, keys[start:end]...).Result()
if err != nil {
return nil, fmt.Errorf("mget: %w", err)
}
values = append(values, got...)
}
// A nil slot is an absent key. A present key holding "" is still an echo
// and still goes, so presence — not the value — decides.
get := func(i int) (string, bool) {
if i >= len(values) || values[i] == nil {
return "", false
}
s, ok := values[i].(string)
return s, ok
}
type candidate struct {
hashID string
hasAnilist bool
echoes []string
}
var cands []candidate
var probe []string // candidates whose anilistid needs their cids/ key-set
for i, h := range hashIDs {
base := i * per
domain, _ := get(base)
fileType, _ := get(base + 1)
if domain != "literature" || fileType != "card" {
continue
}
cand := candidate{hashID: h}
_, cand.hasAnilist = get(base + 2)
for j, f := range literatureCardEchoFields {
if _, ok := get(base + 3 + j); ok {
cand.echoes = append(cand.echoes, f)
}
}
if !cand.hasAnilist && len(cand.echoes) == 0 {
continue
}
if cand.hasAnilist {
probe = append(probe, h)
}
cands = append(cands, cand)
}
// Pass 2: the field-name index of only the cards carrying an `anilistid`
// — the `cids/` members are field NAMES, so the index is where they are
// listed. Pipelined per chunk; still no SCAN.
//
// A card with no index (never touched since before the index existed, and
// missed by the startup WarmFieldIndexes pass) reads as an empty set and
// keeps its `anilistid`: proving it a site card would need exactly the
// per-record SCAN this sweep refuses, and a stamp left in place is the
// safe direction.
siteCard := make(map[string]bool, len(probe))
for start := 0; start < len(probe); start += mgetBatch {
end := start + mgetBatch
if end > len(probe) {
end = len(probe)
}
pipe := c.client.Pipeline()
cmds := make([]*redis.StringSliceCmd, 0, end-start)
for _, h := range probe[start:end] {
cmds = append(cmds, pipe.SMembers(ctx, c.buildFieldsKey(h)))
}
if _, err := pipe.Exec(ctx); err != nil && err != redis.Nil {
return nil, fmt.Errorf("smembers fields: %w", err)
}
for k, cmd := range cmds {
fields, err := cmd.Result()
if err != nil && err != redis.Nil {
return nil, fmt.Errorf("smembers fields %s: %w", probe[start+k], err)
}
siteCard[probe[start+k]] = isSiteCard(fields)
}
}
plans := make([]literatureEchoPlan, 0, len(cands))
for _, cand := range cands {
var fields []string
if cand.hasAnilist && siteCard[cand.hashID] {
fields = append(fields, "anilistid")
}
fields = append(fields, cand.echoes...)
if len(fields) > 0 {
plans = append(plans, literatureEchoPlan{hashID: cand.hashID, fields: fields})
}
}
return plans, nil
}
// SweepLiteratureEchoesWithTimeout is the convenience wrapper used by the API
// handler. Generous timeout: the read phase is batched, but the write phase is
// one round-trip per deleted field.
func (c *Client) SweepLiteratureEchoesWithTimeout(apply bool) (LiteratureEchoSweep, error) {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
defer cancel()
return c.SweepLiteratureEchoes(ctx, apply)
}
@@ -0,0 +1,249 @@
package storage
import (
"context"
"encoding/base32"
"encoding/binary"
"reflect"
"strings"
"testing"
)
// Pinned from meta-read `src/card_cid.rs` — literals, not round-trips.
const (
literatureAnchorCardCID = "bagdsaaaua5qw42lmnfzxi3lbnztwcorrga2tonzy" // ("anilist", "manga:105778")
literatureSiteCardCID = "bagdsaabybrwwc3thmfsgk6bon5zgol3nmfxgoyjpmiygenzsgftgmlldgm4dqljugq4dmllbmeygmlldgjrdaytcgmzdcnjrgi"
// A non-card CID a card may also carry.
literatureContentCID = "bagacbabaec7v3fu2ygzh3e2sybg3fbzmisry2hbtpmck6vx3yftea6vzq35r4"
)
// mintCardCID encodes a card locator the way meta-feeder-sdk
// `compute_card_cid` does. Asserted against the pinned literals before use.
func mintCardCID(source, id string) string {
digest := binary.AppendUvarint(nil, uint64(len(source)))
digest = append(digest, source...)
digest = append(digest, id...)
wire := []byte{0x01}
wire = binary.AppendUvarint(wire, 0x1007)
wire = append(wire, 0x00)
wire = binary.AppendUvarint(wire, uint64(len(digest)))
wire = append(wire, digest...)
return "b" + strings.ToLower(base32.StdEncoding.WithPadding(base32.NoPadding).EncodeToString(wire))
}
func TestIsSiteFormCardLocator(t *testing.T) {
if got := mintCardCID("anilist", "manga:105778"); got != literatureAnchorCardCID {
t.Fatalf("test encoder drifted: %s != %s", got, literatureAnchorCardCID)
}
if got := mintCardCID("mangadex.org", "/manga/b0b721ff-c388-4486-aa0f-c2b0bb321512"); got != literatureSiteCardCID {
t.Fatalf("test encoder drifted: %s != %s", got, literatureSiteCardCID)
}
for _, tc := range []struct {
name string
cid string
want bool
}{
{"anilist anchor (golden)", literatureAnchorCardCID, false},
{"mangadex site (golden)", literatureSiteCardCID, true},
{"openlibrary anchor", mintCardCID("openlibrary", "OL45804W"), false},
{"another site", mintCardCID("weebcentral.com", "/series/01J76XY/chainsaw-man"), true},
// Not an anchor source, but not host-shaped either: meta-read reads it
// as an id-bag key, so it must not be treated as a site.
{"dotless unknown source", mintCardCID("suwayomi", "42"), false},
{"content CID", literatureContentCID, false},
{"title key", "title:chainsaw man", false},
} {
if got := isSiteFormCardLocator(tc.cid); got != tc.want {
t.Errorf("%s: isSiteFormCardLocator = %v, want %v", tc.name, got, tc.want)
}
}
}
// seedLiteratureCorpus writes one record per rule the sweep has to honour.
func seedLiteratureCorpus(t *testing.T, c *Client) {
t.Helper()
// A site card carrying every echo: all of it goes.
seedRecord(t, c, "site-card", map[string]string{
"domain": "literature",
"fileType": "card",
"title": "Chainsaw Man",
"cids/" + literatureSiteCardCID: "true",
"anilistid": "105778",
"workcid": literatureSiteCardCID,
"chapterNumber": "12",
"volumeNumber": "2",
"chapterStart": "1",
"chapterEnd": "20",
})
// An anchor card: its anilistid is its identity and stays; the position
// echo and the workcid still go.
seedRecord(t, c, "anchor-card", map[string]string{
"domain": "literature",
"fileType": "card",
"title": "Chainsaw Man",
"cids/" + literatureAnchorCardCID: "true",
"anilistid": "105778",
"workcid": literatureAnchorCardCID,
"chapterNumber": "3",
})
// A site card that is already clean: nothing planned.
seedRecord(t, c, "clean-site-card", map[string]string{
"domain": "literature",
"fileType": "card",
"cids/" + literatureSiteCardCID: "true",
})
// Both locator forms: which one the id belongs to is unknowable, so it
// stays.
seedRecord(t, c, "merged-card", map[string]string{
"domain": "literature",
"fileType": "card",
"cids/" + literatureSiteCardCID: "true",
"cids/" + literatureAnchorCardCID: "true",
"anilistid": "105778",
})
// No card locator at all: not provably a site card.
seedRecord(t, c, "unlocated-card", map[string]string{
"domain": "literature",
"fileType": "card",
"cids/" + literatureContentCID: "true",
"anilistid": "105778",
})
// A stored empty value is still an echo.
seedRecord(t, c, "blank-echo-card", map[string]string{
"domain": "literature",
"fileType": "card",
"volumeNumber": "",
})
// A chapter is never swept, whatever it carries.
seedRecord(t, c, "chapter", map[string]string{
"domain": "literature",
"fileType": "archive",
"cids/" + literatureContentCID: "true",
"anilistid": "105778",
"workcid": literatureSiteCardCID,
"chapterNumber": "12",
})
// Not literature: out of scope even though it is a card.
seedRecord(t, c, "screen-card", map[string]string{
"domain": "screen",
"fileType": "card",
"workcid": literatureSiteCardCID,
"chapterNumber": "1",
})
}
func TestPlanLiteratureEchoes_ScopesToLiteratureCards(t *testing.T) {
c, _ := newTestClient(t, "")
seedLiteratureCorpus(t, c)
plans, err := c.planLiteratureEchoes(context.Background())
if err != nil {
t.Fatalf("plan: %v", err)
}
got := map[string][]string{}
for _, p := range plans {
got[p.hashID] = p.fields
}
want := map[string][]string{
"site-card": {"anilistid", "workcid", "chapterNumber", "volumeNumber", "chapterStart", "chapterEnd"},
"anchor-card": {"workcid", "chapterNumber"},
"blank-echo-card": {"volumeNumber"},
}
if !reflect.DeepEqual(got, want) {
t.Errorf("plan =\n %v\nwant\n %v", got, want)
}
}
func TestSweepLiteratureEchoes_DryRunThenApply(t *testing.T) {
c, s := newTestClient(t, "mm:")
seedLiteratureCorpus(t, c)
wantByKey := map[string]int{
"anilistid": 1,
"workcid": 2,
"chapterNumber": 2,
"volumeNumber": 2,
"chapterStart": 1,
"chapterEnd": 1,
}
// Dry run: counts, and nothing written.
dry, err := c.SweepLiteratureEchoes(context.Background(), false)
if err != nil {
t.Fatalf("dry run: %v", err)
}
if dry.Planned != 3 || dry.Fixed != 0 {
t.Errorf("dry run planned/fixed = %d/%d, want 3/0", dry.Planned, dry.Fixed)
}
if !reflect.DeepEqual(dry.ByKey, wantByKey) {
t.Errorf("dry run byKey = %v, want %v", dry.ByKey, wantByKey)
}
if m, _ := c.GetMetadataFlat("site-card"); m["anilistid"] != "105778" {
t.Fatalf("dry run deleted anilistid: %v", m)
}
applied, err := c.SweepLiteratureEchoes(context.Background(), true)
if err != nil {
t.Fatalf("apply: %v", err)
}
if applied.Planned != 3 || applied.Fixed != 3 {
t.Errorf("apply planned/fixed = %d/%d, want 3/3", applied.Planned, applied.Fixed)
}
if !reflect.DeepEqual(applied.ByKey, wantByKey) {
t.Errorf("apply byKey = %v, want %v", applied.ByKey, wantByKey)
}
site, err := c.GetMetadataFlat("site-card")
if err != nil {
t.Fatalf("read site-card: %v", err)
}
for _, k := range []string{"anilistid", "workcid", "chapterNumber", "volumeNumber", "chapterStart", "chapterEnd"} {
if _, ok := site[k]; ok {
t.Errorf("site-card still carries %s", k)
}
// The flat key and its field-name index entry must both be gone, or the
// next read would MGET a ghost field.
if s.Exists("mm:file:site-card/" + k) {
t.Errorf("site-card flat key %s survived", k)
}
if member, _ := s.IsMember("mm:file:site-card/"+FieldsField, k); member {
t.Errorf("site-card field index still lists %s", k)
}
}
if site["title"] != "Chainsaw Man" || site["cids/"+literatureSiteCardCID] != "true" {
t.Errorf("site-card lost fields it should keep: %v", site)
}
anchor, _ := c.GetMetadataFlat("anchor-card")
if anchor["anilistid"] != "105778" {
t.Errorf("anchor-card anilistid = %q, want kept", anchor["anilistid"])
}
if _, ok := anchor["workcid"]; ok {
t.Errorf("anchor-card still carries workcid")
}
chapter, _ := c.GetMetadataFlat("chapter")
if chapter["anilistid"] != "105778" || chapter["workcid"] != literatureSiteCardCID || chapter["chapterNumber"] != "12" {
t.Errorf("chapter was touched: %v", chapter)
}
screen, _ := c.GetMetadataFlat("screen-card")
if screen["workcid"] != literatureSiteCardCID {
t.Errorf("screen-card was touched: %v", screen)
}
merged, _ := c.GetMetadataFlat("merged-card")
if merged["anilistid"] != "105778" {
t.Errorf("merged-card anilistid = %q, want kept", merged["anilistid"])
}
// Idempotent: the endpoint is called by hand, and a second call must find
// nothing — with byKey still an empty map, not nil.
again, err := c.SweepLiteratureEchoes(context.Background(), true)
if err != nil {
t.Fatalf("second pass: %v", err)
}
if again.Planned != 0 || again.Fixed != 0 || again.ByKey == nil || len(again.ByKey) != 0 {
t.Errorf("second pass = %+v, want nothing planned and an empty byKey", again)
}
}
+31
View File
@@ -4,6 +4,7 @@ import (
"context"
"fmt"
"log"
"sort"
"strings"
"time"
)
@@ -202,6 +203,11 @@ func (c *Client) GetSearchIndexTuplesFor(hashIDs []string) (tuples []SearchTuple
// blob. Shared by the full and incremental index reads so the two can never
// disagree about what is searchable — a drift there would make a record's
// matchability depend on which path happened to index it.
//
// Beyond the flat fields it folds in every `titles/<lang3>/<name>` member's
// name (METADATA_KEYS.md §3): the name lives in the key and the value is the
// sentinel "true", so a search for an alternative title (`3 percent`) finds the
// record whose `title` is `3%`.
func buildHaystack(fields map[string]string) string {
var sb strings.Builder
for _, f := range searchIndexFields {
@@ -210,5 +216,30 @@ func buildHaystack(fields map[string]string) string {
sb.WriteByte('\n')
}
}
for _, name := range titleMemberNames(fields) {
sb.WriteString(strings.ToLower(name))
sb.WriteByte('\n')
}
return sb.String()
}
// titleMemberNames returns the names of a record's `titles/<lang3>/<name>`
// key-set members, with the `∕` (U+2215) a writer substitutes for `/` mapped
// back. Sorted: the fields map has no order, and the full and incremental
// builds must produce byte-identical haystacks.
func titleMemberNames(fields map[string]string) []string {
var names []string
for k, v := range fields {
rest, ok := strings.CutPrefix(k, "titles/")
if !ok || v == "" || v == "false" {
continue
}
_, name, ok := strings.Cut(rest, "/")
if !ok || name == "" {
continue
}
names = append(names, strings.ReplaceAll(name, "\u2215", "/"))
}
sort.Strings(names)
return names
}
+19
View File
@@ -174,6 +174,8 @@ func TestDirty_IncrementalTuplesMatchTheFullRebuild(t *testing.T) {
"title": "Parity Release",
"fileName": "parity.s01e01.mkv",
"fileType": "video",
"titles/eng/3 Percent": "true",
"titles/fra/3 Pourcent": "true",
"categories/XXX": "true",
"categories/newznab/5070": "true",
}); err != nil {
@@ -209,3 +211,20 @@ func TestDirty_IncrementalTuplesMatchTheFullRebuild(t *testing.T) {
}
})
}
// A `titles/<lang3>/<name>` member contributes its NAME (the key), never its
// "true" value, and the `∕` a writer substituted for `/` reads back as `/`.
func TestBuildHaystack_FoldsTitleMemberNames(t *testing.T) {
got := buildHaystack(map[string]string{
"title": "3%",
"titles/fra/3 Pourcent": "true",
"titles/eng/3 Percent": "true",
"titles/und/Fate\u2215Zero": "true",
"titles/eng": "Retired Scalar",
"titles/eng/": "true",
})
want := "3%\n3 percent\n3 pourcent\nfate/zero\n"
if got != want {
t.Errorf("haystack = %q, want %q", got, want)
}
}
+180 -1
View File
@@ -17,6 +17,7 @@ package storage
// udl:rec:<uid>/<cid>/<key> HASH { record, version, ts, value? }
// udl:idx:user:<uid>:key:<key> SET of <cid> -> active_for_user_key (My List, Continue Watching)
// udl:idx:user:<uid>:cid:<cid> SET of <key> -> list_for_user_cid
// udl:idx:user:<uid>:cids SET of <cid> -> whole-profile export (recovery)
// udl:idx:cid:<cid>:key:<key> SET of <uid> -> aggregate
// udl:idx:cid:<cid>:users SET of <uid> -> uids_for_cid
//
@@ -27,7 +28,9 @@ package storage
import (
"context"
"fmt"
"sort"
"strconv"
"strings"
"time"
"github.com/redis/go-redis/v9"
@@ -66,11 +69,24 @@ func udlIdxUserCid(uid, cid string) string { return fmt.Sprintf("udl:idx:user:%s
func udlIdxCidKey(cid, key string) string { return fmt.Sprintf("udl:idx:cid:%s:key:%s", cid, key) }
func udlIdxCidUsers(cid string) string { return fmt.Sprintf("udl:idx:cid:%s:users", cid) }
// udlIdxUserCids is the set of every cid a uid has ANY record for — the one
// index that makes "export this whole profile" a set read instead of a
// keyspace scan.
//
// The other four indexes each answer a question with one coordinate already
// known (a key, a cid). Cross-instance recovery knows neither: it has a uid and
// nothing else. Without this set the only ways to enumerate a profile are to
// iterate the hardcoded key list — which silently misses any key added later —
// or to SCAN `udl:rec:<uid>/*`, which is the full-keyspace-read pathology that
// already made `/api/udl/users/stats` admin-only.
func udlIdxUserCids(uid string) string { return fmt.Sprintf("udl:idx:user:%s:cids", uid) }
// udlUpsertScript performs the version-gated upsert + index maintenance
// atomically. Accepts (returns 1) only when the incoming version strictly
// exceeds the stored one; otherwise returns 0 (stale).
//
// KEYS: 1=cell 2=idx:user:key 3=idx:user:cid 4=idx:cid:key 5=idx:cid:users
// 6=idx:user:cids
// ARGV: 1=version 2=ts 3=record 4=value 5=hasValue("1"/"0")
// 6=cid 7=key 8=uid 9=tombstone("1"/"0")
//
@@ -78,9 +94,16 @@ func udlIdxCidUsers(cid string) string { return fmt.Sprintf("udl:idx:cid:%s
// would let an older write resurrect the value — but it SREMs the cid from the
// user+key index. That index is what `active_for_user_key` returns, so leaving
// deleted cids in it means every My List / Continue Watching read carries every
// title ever un-liked or finished, forever. The other three indexes are left
// title ever un-liked or finished, forever. The other four indexes are left
// alone: they answer "who has an opinion about this cid", where a tombstone is
// still an answer.
//
// ⚠ `idx:user:cids` (KEYS[6]) in particular must NEVER be pruned on a
// tombstone, even though it looks symmetric with KEYS[2]. It is the export
// index: a cid dropped from it is a cid whose tombstone never travels, and a
// tombstone that does not travel is exactly how an un-liked title comes back
// from the dead on the other box — the peer still holds the older `like` at a
// lower version, and with no tombstone to beat it, nothing ever supersedes it.
var udlUpsertScript = redis.NewScript(`
local stored = redis.call('HGET', KEYS[1], 'version')
if stored and tonumber(stored) >= tonumber(ARGV[1]) then
@@ -100,6 +123,7 @@ end
redis.call('SADD', KEYS[3], ARGV[7])
redis.call('SADD', KEYS[4], ARGV[8])
redis.call('SADD', KEYS[5], ARGV[8])
redis.call('SADD', KEYS[6], ARGV[6])
return 1
`)
@@ -126,6 +150,7 @@ func (c *Client) UDLUpsertIfNewer(uid, cid, key string, version, ts int64, recor
c.buildKey(udlIdxUserCid(uid, cid)),
c.buildKey(udlIdxCidKey(cid, key)),
c.buildKey(udlIdxCidUsers(cid)),
c.buildKey(udlIdxUserCids(uid)),
}
hv := "0"
if hasValue {
@@ -250,6 +275,160 @@ func (c *Client) UDLListUserCid(uid, cid string) ([]UDLUserCidEntry, error) {
return out, nil
}
// UDLUserAllEntry is one row of a whole-profile export.
type UDLUserAllEntry struct {
Cid string `json:"cid"`
Key string `json:"key"`
Record string `json:"record"`
Version int64 `json:"version"`
Ts int64 `json:"ts"`
}
// UDLAllForUser returns every cell a uid owns, for cross-instance recovery.
//
// Ordered by cid (sorted, so paging is stable across calls even though the
// underlying index is a Redis SET), starting strictly after `afterCid`. A cid's
// keys are never split across a page: the page ends on a cid boundary once
// `limit` rows have been emitted, and `nextCid` is the last cid included — pass
// it back as `afterCid` to continue. `nextCid` is empty when the export is
// complete.
//
// Tombstones are included. They are the CRDT state that stops a deleted value
// resurrecting on the box that receives this export, so omitting them would
// make recovery actively wrong rather than merely incomplete.
func (c *Client) UDLAllForUser(uid, afterCid string, limit int) ([]UDLUserAllEntry, string, error) {
c.mu.RLock()
defer c.mu.RUnlock()
if c.client == nil {
return nil, "", fmt.Errorf("not connected")
}
if limit <= 0 {
limit = 1000
}
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
cids, err := c.client.SMembers(ctx, c.buildKey(udlIdxUserCids(uid))).Result()
if err != nil {
return nil, "", fmt.Errorf("smembers failed: %w", err)
}
sort.Strings(cids)
out := make([]UDLUserAllEntry, 0, limit)
nextCid := ""
for _, cid := range cids {
if afterCid != "" && cid <= afterCid {
continue
}
if len(out) >= limit {
nextCid = out[len(out)-1].Cid
break
}
keys, err := c.client.SMembers(ctx, c.buildKey(udlIdxUserCid(uid, cid))).Result()
if err != nil {
return nil, "", fmt.Errorf("smembers failed: %w", err)
}
sort.Strings(keys)
for _, key := range keys {
cell, ok, err := c.udlReadCell(ctx, udlRecKey(uid, cid, key))
if err != nil {
return nil, "", err
}
if !ok {
continue
}
out = append(out, UDLUserAllEntry{
Cid: cid, Key: key, Record: cell.Record, Version: cell.Version, Ts: cell.Ts,
})
}
}
return out, nextCid, nil
}
// UDLCountUserCids is the number of distinct cids a uid holds records for.
//
// O(1) (a Redis SCARD), which is what makes it usable as a per-push quota
// check on the gateway mirror — a quota that cost a keyspace walk would be a
// denial-of-service in its own right. Counts cids rather than cells because
// that is what the index holds; a profile is bounded by the titles it has
// touched, not by how many keys it has per title.
func (c *Client) UDLCountUserCids(uid string) (int64, error) {
c.mu.RLock()
defer c.mu.RUnlock()
if c.client == nil {
return 0, fmt.Errorf("not connected")
}
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
n, err := c.client.SCard(ctx, c.buildKey(udlIdxUserCids(uid))).Result()
if err != nil {
return 0, fmt.Errorf("scard failed: %w", err)
}
return n, nil
}
// UDLBackfillUserCids rebuilds the `idx:user:<uid>:cids` sets for data written
// before that index existed, and reports how many cids it added.
//
// One-shot: the upsert script maintains the set from now on. It is a SCAN, but
// a targeted one (`udl:idx:user:*:cid:*`, which is one key per (uid,cid), not
// per record) run once at boot rather than per read — the distinction that
// makes the difference between a migration and the pathology this index exists
// to avoid.
func (c *Client) UDLBackfillUserCids() (int, error) {
c.mu.Lock()
defer c.mu.Unlock()
if c.client == nil {
return 0, fmt.Errorf("not connected")
}
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
defer cancel()
pattern := c.buildKey("udl:idx:user:*:cid:*")
prefix := c.buildKey("udl:idx:user:")
added := 0
var cursor uint64
for {
keys, next, err := c.client.Scan(ctx, cursor, pattern, 500).Result()
if err != nil {
return added, fmt.Errorf("scan failed: %w", err)
}
for _, k := range keys {
// <prefix><uid>:cid:<cid> — split on the FIRST separator, not the
// last. A uid is multibase base58btc, whose alphabet contains no
// colon at all, so the first ":cid:" always ends the uid. A cid
// that happens to contain ":cid:" in its own name would make the
// last occurrence split in the wrong place.
rest := strings.TrimPrefix(k, prefix)
i := strings.Index(rest, ":cid:")
if i < 0 {
continue
}
uid, cid := rest[:i], rest[i+len(":cid:"):]
if uid == "" || cid == "" {
continue
}
n, err := c.client.SAdd(ctx, c.buildKey(udlIdxUserCids(uid)), cid).Result()
if err != nil {
return added, fmt.Errorf("sadd failed: %w", err)
}
added += int(n)
}
cursor = next
if cursor == 0 {
break
}
}
return added, nil
}
// UDLCidUsers returns every uid that has any record for cid — uids_for_cid.
func (c *Client) UDLCidUsers(cid string) ([]string, error) {
c.mu.RLock()
+5 -3
View File
@@ -179,7 +179,7 @@ func (c *Client) UDLAllUserStats() (map[string]UDLUserStats, error) {
}
// UDLPurgeUser removes every trace of one account from the User Data Layer:
// its cells, its two user-scoped indexes, and its membership in the shared
// its cells, its three user-scoped indexes, and its membership in the shared
// cid-scoped indexes. Returns what was removed.
//
// The shared indexes are the part that is easy to miss and expensive to leave.
@@ -225,8 +225,10 @@ func (c *Client) UDLPurgeUser(uid string) (UDLUserStats, error) {
stats.Cids = len(distinctCids)
stats.Keys = len(distinctKeys)
// The two user-scoped indexes are enumerable directly, and are dropped
// whole rather than member-by-member. Scanning them separately also catches
// The three user-scoped indexes (`:key:*`, `:cid:*`, `:cids`) are enumerable
// directly, and are dropped whole rather than member-by-member. The glob is
// what keeps this correct as indexes are added — `:cids` needed no change
// here when it landed. Scanning them separately also catches
// index entries whose cell is already gone — exactly the state a previously
// half-finished purge leaves behind.
idxPrefix := c.buildKey("udl:idx:user:") + uid + ":"
+234
View File
@@ -0,0 +1,234 @@
package storage
import (
"context"
"testing"
)
// The whole-profile export is what cross-instance recovery reads: a box that
// has just imported a secret key knows the uid and nothing else — not which
// cids the profile touched, not which keys. Every one of these tests is about
// that one caller.
func TestUDLAllForUser_ReturnsEveryCellAcrossCidsAndKeys(t *testing.T) {
c, _ := newTestClient(t, "")
const uid = "zAlice"
writes := []struct{ cid, key, rec string }{
{"bagcsaaaaone", "like", "rec-one-like"},
{"bagcsaaaaone", "seek", "rec-one-seek"},
{"bagcsaaaatwo", "rating", "rec-two-rating"},
{"bagcsaaaatwo", "profile:name", "rec-two-name"},
{"bagcsaaathree", "watched", "rec-three-watched"},
}
for _, w := range writes {
if ok, err := c.UDLUpsertIfNewer(uid, w.cid, w.key, 1, 100, w.rec, "", false, false); err != nil || !ok {
t.Fatalf("upsert %s/%s: ok=%v err=%v", w.cid, w.key, ok, err)
}
}
entries, next, err := c.UDLAllForUser(uid, "", 0)
if err != nil {
t.Fatalf("export: %v", err)
}
if next != "" {
t.Fatalf("export of 5 rows should be complete, got nextCid=%q", next)
}
if len(entries) != len(writes) {
t.Fatalf("want %d rows, got %d: %+v", len(writes), len(entries), entries)
}
seen := map[string]string{}
for _, e := range entries {
seen[e.Cid+"/"+e.Key] = e.Record
}
for _, w := range writes {
if got := seen[w.cid+"/"+w.key]; got != w.rec {
t.Errorf("%s/%s: want record %q, got %q", w.cid, w.key, w.rec, got)
}
}
}
// The failure this guards is silent and total: a uid whose cids were never
// indexed exports as empty, and recovery reports success having restored
// nothing.
func TestUDLAllForUser_IsEmptyForAnUnknownUid(t *testing.T) {
c, _ := newTestClient(t, "")
if ok, err := c.UDLUpsertIfNewer("zAlice", "bagcsaaaaone", "like", 1, 100, "rec", "", false, false); err != nil || !ok {
t.Fatalf("upsert: ok=%v err=%v", ok, err)
}
entries, _, err := c.UDLAllForUser("zBob", "", 0)
if err != nil {
t.Fatalf("export: %v", err)
}
if len(entries) != 0 {
t.Fatalf("Bob owns nothing; got %+v", entries)
}
}
// ⚠ The tombstone rule, and the reason the cids index is never pruned.
//
// A tombstone must EXPORT. It is the only thing that can beat the older live
// value the other box still holds at a lower version — drop it from the export
// and an un-liked title comes back from the dead on every peer that already
// saw the `like`. This is the one place the cids index deliberately diverges
// from the user+key index, which does prune.
func TestUDLAllForUser_ExportsTombstones(t *testing.T) {
c, _ := newTestClient(t, "")
const (
uid = "zAlice"
cid = "bagcsaaaaone"
key = "like"
)
if ok, err := c.UDLUpsertIfNewer(uid, cid, key, 1, 100, "rec-liked", "", false, false); err != nil || !ok {
t.Fatalf("live upsert: ok=%v err=%v", ok, err)
}
if ok, err := c.UDLUpsertIfNewer(uid, cid, key, 2, 200, "rec-unliked", "", false, true); err != nil || !ok {
t.Fatalf("tombstone upsert: ok=%v err=%v", ok, err)
}
// The user+key index prunes it — that is what keeps My List short.
live, err := c.UDLListUserKey(uid, key)
if err != nil {
t.Fatalf("active_for_user_key: %v", err)
}
if len(live) != 0 {
t.Fatalf("tombstoned cid should leave the user+key index, got %+v", live)
}
// The export must still carry it, at the tombstone's version.
entries, _, err := c.UDLAllForUser(uid, "", 0)
if err != nil {
t.Fatalf("export: %v", err)
}
if len(entries) != 1 {
t.Fatalf("tombstone must export, got %+v", entries)
}
if entries[0].Record != "rec-unliked" || entries[0].Version != 2 {
t.Fatalf("export must carry the tombstone, not the superseded value: %+v", entries[0])
}
}
func TestUDLAllForUser_PagesOnACidBoundaryAndResumes(t *testing.T) {
c, _ := newTestClient(t, "")
const uid = "zAlice"
cids := []string{"bagcsaaaaa1", "bagcsaaaaa2", "bagcsaaaaa3"}
for _, cid := range cids {
for _, key := range []string{"like", "seek"} {
if ok, err := c.UDLUpsertIfNewer(uid, cid, key, 1, 100, "rec-"+cid+"-"+key, "", false, false); err != nil || !ok {
t.Fatalf("upsert: ok=%v err=%v", ok, err)
}
}
}
// A cid's keys are never split across a page, so a limit of 1 still yields
// both of the first cid's rows.
page1, next1, err := c.UDLAllForUser(uid, "", 1)
if err != nil {
t.Fatalf("page 1: %v", err)
}
if len(page1) != 2 || page1[0].Cid != cids[0] {
t.Fatalf("page 1 should hold both rows of %s, got %+v", cids[0], page1)
}
if next1 != cids[0] {
t.Fatalf("cursor should name the last cid emitted, want %q got %q", cids[0], next1)
}
// Walking the cursor must visit every row exactly once.
all := append([]UDLUserAllEntry{}, page1...)
cursor := next1
for cursor != "" {
page, next, err := c.UDLAllForUser(uid, cursor, 1)
if err != nil {
t.Fatalf("page after %q: %v", cursor, err)
}
all = append(all, page...)
cursor = next
}
if len(all) != 6 {
t.Fatalf("want 6 rows across all pages, got %d: %+v", len(all), all)
}
seen := map[string]bool{}
for _, e := range all {
k := e.Cid + "/" + e.Key
if seen[k] {
t.Fatalf("%s emitted twice across pages", k)
}
seen[k] = true
}
}
// Data written before the cids index existed must still export, or every
// profile that predates this feature silently recovers as empty.
func TestUDLBackfillUserCids_MakesLegacyDataExportable(t *testing.T) {
c, _ := newTestClient(t, "")
const (
uid = "zAlice"
cid = "bagcsaaaaone"
key = "like"
)
if ok, err := c.UDLUpsertIfNewer(uid, cid, key, 1, 100, "rec", "", false, false); err != nil || !ok {
t.Fatalf("upsert: ok=%v err=%v", ok, err)
}
// Simulate the pre-index state: the cell and the other indexes exist, the
// cids set does not.
ctx := context.Background()
if err := c.client.Del(ctx, c.buildKey(udlIdxUserCids(uid))).Err(); err != nil {
t.Fatalf("del cids index: %v", err)
}
if entries, _, err := c.UDLAllForUser(uid, "", 0); err != nil || len(entries) != 0 {
t.Fatalf("precondition: legacy data should export as empty, got %+v err=%v", entries, err)
}
added, err := c.UDLBackfillUserCids()
if err != nil {
t.Fatalf("backfill: %v", err)
}
if added != 1 {
t.Fatalf("want 1 cid backfilled, got %d", added)
}
entries, _, err := c.UDLAllForUser(uid, "", 0)
if err != nil {
t.Fatalf("export after backfill: %v", err)
}
if len(entries) != 1 || entries[0].Cid != cid || entries[0].Key != key {
t.Fatalf("backfilled data should export, got %+v", entries)
}
// Idempotent: a second run adds nothing.
if added, err := c.UDLBackfillUserCids(); err != nil || added != 0 {
t.Fatalf("second backfill should be a no-op, added=%d err=%v", added, err)
}
}
// The backfill splits `<uid>:cid:<cid>` on the FIRST separator, because a uid
// is multibase base58btc and its alphabet has no colon. Splitting on the last
// occurrence — the obvious-looking choice — mis-parses a cid that contains
// ":cid:" in its own name into uid "zAlice:cid:weird", cid "name".
func TestUDLBackfillUserCids_SplitsOnTheFirstSeparator(t *testing.T) {
c, _ := newTestClient(t, "")
const uid = "zAlice"
cid := "weird:cid:name"
if ok, err := c.UDLUpsertIfNewer(uid, cid, "like", 1, 100, "rec", "", false, false); err != nil || !ok {
t.Fatalf("upsert: ok=%v err=%v", ok, err)
}
ctx := context.Background()
if err := c.client.Del(ctx, c.buildKey(udlIdxUserCids(uid))).Err(); err != nil {
t.Fatalf("del cids index: %v", err)
}
if _, err := c.UDLBackfillUserCids(); err != nil {
t.Fatalf("backfill: %v", err)
}
entries, _, err := c.UDLAllForUser(uid, "", 0)
if err != nil {
t.Fatalf("export: %v", err)
}
if len(entries) != 1 || entries[0].Cid != cid {
t.Fatalf("want the cid parsed whole, got %+v", entries)
}
}
+12 -4
View File
@@ -50,8 +50,18 @@ func (d *Dispatcher) dispatchToStream(event FileEvent) {
client := d.storageClient
d.mu.RUnlock()
if err := PublishEvent(client, event); err != nil {
log.Printf("[Dispatcher] Failed to publish to stream: %v", err)
}
}
// PublishEvent appends one file event to the file:events stream. Exported so a
// caller with no Dispatcher (POST /api/files/register on a box whose watcher is
// off) emits exactly the entry the watcher would — meta-share's ipfs_seed
// consumes this stream to chunk, seed and provide new files.
func PublishEvent(client *storage.Client, event FileEvent) error {
if client == nil || !client.IsConnected() {
return
return fmt.Errorf("storage client not connected")
}
// Build stream fields
@@ -79,9 +89,7 @@ func (d *Dispatcher) dispatchToStream(event FileEvent) {
// `EmitReset → ClearStream` path is still the only way the stream gets
// truncated to zero (signalled to consumers via the `reset` event).
_, err := client.XAdd(EventsStream, EventsStreamMaxLen, fields)
if err != nil {
log.Printf("[Dispatcher] Failed to publish to stream: %v", err)
}
return err
}
// EmitReset clears the stream and emits a reset event
+2 -37
View File
@@ -4,7 +4,6 @@ import (
"log"
"os"
"path/filepath"
"strconv"
"strings"
"sync"
@@ -54,42 +53,8 @@ func (w *Watcher) writeFileTuple(absPath string, size, mtimeNano int64, midhash
return
}
uuid, err := w.storage.GetByCID(midhash)
if err != nil {
log.Printf("[Watcher] GetByCID(%s): %v", midhash, err)
return
}
if uuid == "" {
// New content. Mint sets filePath/sizeByte/mtimeNano; SetProperty
// then writes the midhash as a cids/<cid> key-set member — and the
// cid_resolution.go hook registers the reverse-index alias for us.
uuid, err = w.storage.Mint(absPath, size, mtimeNano)
if err != nil {
log.Printf("[Watcher] Mint(%s): %v", absPath, err)
return
}
if err := w.storage.SetProperty(uuid, storage.CIDsKeyPrefix+midhash, "true"); err != nil {
log.Printf("[Watcher] SetProperty cids/%s for %s: %v", midhash, uuid, err)
}
return
}
// Known content. If this is a new physical path for the same content,
// record it as a duplicate. Otherwise refresh size/mtime — same content,
// same path, but the file may have been rewritten with a fresh mtime.
existing, _ := w.storage.GetProperty(uuid, "filePath")
if existing != absPath {
if _, err := w.storage.AddDuplicatePath(uuid, absPath); err != nil {
log.Printf("[Watcher] AddDuplicatePath %s @ %s: %v", uuid, absPath, err)
}
return
}
if _, err := w.storage.MergeMetadataFlat(uuid, map[string]string{
"sizeByte": strconv.FormatInt(size, 10),
"mtimeNano": strconv.FormatInt(mtimeNano, 10),
}); err != nil {
log.Printf("[Watcher] refresh tuple for %s: %v", uuid, err)
if _, _, err := w.storage.RegisterFileTuple(absPath, size, mtimeNano, midhash); err != nil {
log.Printf("[Watcher] register %s: %v", absPath, err)
}
}
-6
View File
@@ -77,15 +77,9 @@ func TestConfigPaths(t *testing.T) {
t.Errorf("Unexpected LockFilePath: %s", cfg.LockFilePath())
}
if cfg.InfoFilePath() != "/test/locks/kv-leader.info" {
t.Errorf("Unexpected InfoFilePath: %s", cfg.InfoFilePath())
}
if cfg.RedisDataDir() != "/test/db/redis" {
t.Errorf("Unexpected RedisDataDir: %s", cfg.RedisDataDir())
}
if cfg.ServicesDir() != "/test/services" {
t.Errorf("Unexpected ServicesDir: %s", cfg.ServicesDir())
}
}
-144
View File
@@ -1,144 +0,0 @@
package test
import (
"encoding/json"
"os"
"path/filepath"
"testing"
"github.com/metazla/meta-core/internal/leader"
)
func TestLeaderLockInfoJSON(t *testing.T) {
info := &leader.LeaderLockInfo{
Host: "test-host",
API: "redis://10.0.0.1:6379",
HTTP: "http://10.0.0.1:8180",
BaseURL: "http://localhost:8180",
Timestamp: 1704067200000,
PID: 12345,
}
// Marshal to JSON
data, err := json.MarshalIndent(info, "", " ")
if err != nil {
t.Fatalf("Failed to marshal: %v", err)
}
// Verify JSON structure matches TypeScript format
var parsed map[string]interface{}
if err := json.Unmarshal(data, &parsed); err != nil {
t.Fatalf("Failed to unmarshal: %v", err)
}
if parsed["host"] != "test-host" {
t.Errorf("Expected host 'test-host', got '%v'", parsed["host"])
}
if parsed["api"] != "redis://10.0.0.1:6379" {
t.Errorf("Expected api 'redis://10.0.0.1:6379', got '%v'", parsed["api"])
}
if parsed["http"] != "http://10.0.0.1:8180" {
t.Errorf("Expected http 'http://10.0.0.1:8180', got '%v'", parsed["http"])
}
if parsed["baseUrl"] != "http://localhost:8180" {
t.Errorf("Expected baseUrl 'http://localhost:8180', got '%v'", parsed["baseUrl"])
}
// Timestamp should be a number
if _, ok := parsed["timestamp"].(float64); !ok {
t.Errorf("Expected timestamp to be a number, got %T", parsed["timestamp"])
}
// PID should be a number
if _, ok := parsed["pid"].(float64); !ok {
t.Errorf("Expected pid to be a number, got %T", parsed["pid"])
}
}
func TestLeaderLockInfoJSONOmitEmpty(t *testing.T) {
info := &leader.LeaderLockInfo{
Host: "test-host",
API: "redis://10.0.0.1:6379",
HTTP: "http://10.0.0.1:8180",
BaseURL: "", // Empty - should be omitted
Timestamp: 1704067200000,
PID: 12345,
}
data, err := json.Marshal(info)
if err != nil {
t.Fatalf("Failed to marshal: %v", err)
}
// baseUrl should not be present when empty
var parsed map[string]interface{}
if err := json.Unmarshal(data, &parsed); err != nil {
t.Fatalf("Failed to unmarshal: %v", err)
}
if _, exists := parsed["baseUrl"]; exists {
t.Error("Expected baseUrl to be omitted when empty")
}
}
func TestLeaderInfoFileFormat(t *testing.T) {
// Create a temp directory
tmpDir, err := os.MkdirTemp("", "meta-core-test")
if err != nil {
t.Fatalf("Failed to create temp dir: %v", err)
}
defer os.RemoveAll(tmpDir)
info := &leader.LeaderLockInfo{
Host: "test-host",
API: "redis://10.0.0.1:6379",
HTTP: "http://10.0.0.1:8180",
Timestamp: 1704067200000,
PID: 12345,
}
// Write to file
infoPath := filepath.Join(tmpDir, "kv-leader.info")
data, err := json.MarshalIndent(info, "", " ")
if err != nil {
t.Fatalf("Failed to marshal: %v", err)
}
if err := os.WriteFile(infoPath, data, 0644); err != nil {
t.Fatalf("Failed to write file: %v", err)
}
// Read and parse
readData, err := os.ReadFile(infoPath)
if err != nil {
t.Fatalf("Failed to read file: %v", err)
}
var parsed leader.LeaderLockInfo
if err := json.Unmarshal(readData, &parsed); err != nil {
t.Fatalf("Failed to unmarshal: %v", err)
}
if parsed.Host != info.Host {
t.Errorf("Host mismatch: expected '%s', got '%s'", info.Host, parsed.Host)
}
if parsed.API != info.API {
t.Errorf("API mismatch: expected '%s', got '%s'", info.API, parsed.API)
}
if parsed.HTTP != info.HTTP {
t.Errorf("HTTP mismatch: expected '%s', got '%s'", info.HTTP, parsed.HTTP)
}
if parsed.Timestamp != info.Timestamp {
t.Errorf("Timestamp mismatch: expected %d, got %d", info.Timestamp, parsed.Timestamp)
}
if parsed.PID != info.PID {
t.Errorf("PID mismatch: expected %d, got %d", info.PID, parsed.PID)
}
}