sendnrw/decent-webui
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

create_aio #1

Merged
groot merged 1 commits from aio into main 2025-09-27 07:35:23 +00:00
Conversation 0 Commits 1 Files Changed 13 +1387 -907
13 changed files with 1387 additions and 907 deletions
Download Patch File Download Diff File Expand all files Collapse all files

363
cmd/filesvc/main.go
View File

@@ -1,363 +0,0 @@
package main
import (
"context"
"crypto/subtle"
"embed"
"encoding/json"
"errors"
"log"
"mime"
"net/http"
"os"
"os/signal"
"path/filepath"
"strconv"
"strings"
"syscall"
"time"
"git.send.nrw/sendnrw/decent-webui/internal/store"
)
//go:embed ui/*
var uiFS embed.FS
type Config struct {
ListenAddr string
DataDir string
APIKey string
}
func (c Config) BlobDir() string { return filepath.Join(c.DataDir, "blobs") }
func (c Config) MetaDir() string { return filepath.Join(c.DataDir, "meta") }
func (c Config) TempDir() string { return filepath.Join(c.DataDir, "tmp") }
func getenv(k, d string) string {
if v := os.Getenv(k); v != "" {
return v
}
return d
}
func LoadConfig() Config {
addr := getenv("FILESVC_LISTEN", ":8085")
datadir := getenv("FILESVC_DATA", "/data")
key := os.Getenv("FILESVC_API_KEY")
if key == "" {
log.Println("[warn] FILESVC_API_KEY is empty — set it for protection")
}
return Config{ListenAddr: addr, DataDir: datadir, APIKey: key}
}
type App struct {
cfg Config
store *store.Store
}
func main() {
cfg := LoadConfig()
for _, p := range []string{cfg.DataDir, cfg.BlobDir(), cfg.MetaDir(), cfg.TempDir()} {
if err := os.MkdirAll(p, 0o755); err != nil {
log.Fatalf("mkdir %s: %v", p, err)
}
}
st, err := store.Open(cfg.BlobDir(), cfg.MetaDir(), cfg.TempDir())
if err != nil {
log.Fatal(err)
}
app := &App{cfg: cfg, store: st}
mux := http.NewServeMux()
// API routes
mux.HandleFunc("/healthz", app.health)
mux.HandleFunc("/v1/files", app.with(app.files))
mux.HandleFunc("/v1/files/", app.with(app.fileByID)) // /v1/files/{id}[ /meta]
mux.HandleFunc("/v1/uploads", app.with(app.uploadsRoot)) // POST init
mux.HandleFunc("/v1/uploads/", app.with(app.uploadsByID)) // parts/complete/abort
// UI routes (embedded)
mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/" {
http.NotFound(w, r)
return
}
http.ServeFileFS(w, r, uiFS, "ui/index.html")
})
mux.Handle("/static/", http.StripPrefix("/static/", http.FileServerFS(uiFS)))
srv := &http.Server{
Addr: cfg.ListenAddr,
Handler: logMiddleware(securityHeaders(mux)),
ReadTimeout: 60 * time.Second,
ReadHeaderTimeout: 10 * time.Second,
WriteTimeout: 0,
IdleTimeout: 120 * time.Second,
}
go func() {
log.Printf("file-service listening on %s", cfg.ListenAddr)
if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
log.Fatalf("server: %v", err)
}
}()
stop := make(chan os.Signal, 1)
signal.Notify(stop, os.Interrupt, syscall.SIGTERM)
<-stop
log.Println("shutting down...")
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
_ = srv.Shutdown(ctx)
}
func (a *App) with(h func(http.ResponseWriter, *http.Request)) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
if a.cfg.APIKey != "" {
key := r.Header.Get("X-API-Key")
if subtle.ConstantTimeCompare([]byte(key), []byte(a.cfg.APIKey)) != 1 {
http.Error(w, "unauthorized", http.StatusUnauthorized)
return
}
}
h(w, r)
}
}
func logMiddleware(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
start := time.Now()
next.ServeHTTP(w, r)
log.Printf("%s %s %s", r.Method, r.URL.Path, time.Since(start))
})
}
func securityHeaders(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
// immer sinnvolle Sicherheits-Header
w.Header().Set("X-Content-Type-Options", "nosniff")
w.Header().Set("X-Frame-Options", "DENY")
w.Header().Set("Referrer-Policy", "no-referrer")
// Für UI (/, /static/...) dürfen CSS/JS & XHR von "self" laden.
if r.URL.Path == "/" || strings.HasPrefix(r.URL.Path, "/static/") {
w.Header().Set("Content-Security-Policy",
"default-src 'self'; script-src 'self'; style-src 'self'; img-src 'self' data:; connect-src 'self'; object-src 'none'; base-uri 'self'; frame-ancestors 'none'")
} else {
// Für API schön streng
w.Header().Set("Content-Security-Policy",
"default-src 'none'; object-src 'none'; base-uri 'none'; frame-ancestors 'none'")
}
next.ServeHTTP(w, r)
})
}
func (a *App) writeJSON(w http.ResponseWriter, status int, v any) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(status)
_ = json.NewEncoder(w).Encode(v)
}
func (a *App) health(w http.ResponseWriter, _ *http.Request) {
w.WriteHeader(200)
_, _ = w.Write([]byte("ok"))
}
// --- Routes ---
// /v1/files (GET list, POST upload)
func (a *App) files(w http.ResponseWriter, r *http.Request) {
switch r.Method {
case http.MethodGet:
q := r.URL.Query().Get("q")
off := atoiDefault(r.URL.Query().Get("offset"), 0)
lim := atoiDefault(r.URL.Query().Get("limit"), 50)
items, next, err := a.store.List(r.Context(), q, off, lim)
if err != nil {
http.Error(w, err.Error(), 500)
return
}
a.writeJSON(w, 200, map[string]any{"items": items, "next": next})
case http.MethodPost:
r.Body = http.MaxBytesReader(w, r.Body, 1<<34) // ~16GiB
ct := r.Header.Get("Content-Type")
name := r.Header.Get("X-Filename")
meta := r.URL.Query().Get("meta")
if strings.HasPrefix(ct, "multipart/") {
if err := r.ParseMultipartForm(32 << 20); err != nil {
http.Error(w, err.Error(), 400)
return
}
f, hdr, err := r.FormFile("file")
if err != nil {
http.Error(w, err.Error(), 400)
return
}
defer f.Close()
if hdr != nil {
name = hdr.Filename
}
rec, err := a.store.Put(r.Context(), f, name, meta)
if err != nil {
http.Error(w, err.Error(), 500)
return
}
a.writeJSON(w, 201, rec)
return
}
rec, err := a.store.Put(r.Context(), r.Body, name, meta)
if err != nil {
http.Error(w, err.Error(), 500)
return
}
a.writeJSON(w, 201, rec)
default:
w.WriteHeader(http.StatusMethodNotAllowed)
}
}
// /v1/files/{id} and /v1/files/{id}/meta
func (a *App) fileByID(w http.ResponseWriter, r *http.Request) {
// path after /v1/files/
rest := strings.TrimPrefix(r.URL.Path, "/v1/files/")
parts := strings.Split(rest, "/")
if len(parts) == 0 || parts[0] == "" {
http.NotFound(w, r)
return
}
id := parts[0]
if len(parts) == 2 && parts[1] == "meta" {
switch r.Method {
case http.MethodGet:
rec, err := a.store.GetMeta(r.Context(), id)
if err != nil {
http.Error(w, err.Error(), 404)
return
}
a.writeJSON(w, 200, rec)
case http.MethodPut:
var m map[string]string
if err := json.NewDecoder(r.Body).Decode(&m); err != nil {
http.Error(w, err.Error(), 400)
return
}
rec, err := a.store.UpdateMeta(r.Context(), id, m)
if err != nil {
http.Error(w, err.Error(), 500)
return
}
a.writeJSON(w, 200, rec)
default:
w.WriteHeader(http.StatusMethodNotAllowed)
}
return
}
// /v1/files/{id}
switch r.Method {
case http.MethodGet:
f, rec, err := a.store.Open(r.Context(), id)
if err != nil {
http.Error(w, err.Error(), 404)
return
}
defer f.Close()
ctype := rec.ContentType
if ctype == "" {
ctype = mime.TypeByExtension(filepath.Ext(rec.Name))
}
if ctype == "" {
ctype = "application/octet-stream"
}
w.Header().Set("Content-Type", ctype)
w.Header().Set("Content-Length", strconv.FormatInt(rec.Size, 10))
w.Header().Set("Accept-Ranges", "bytes")
if r.URL.Query().Get("download") == "1" {
w.Header().Set("Content-Disposition", "attachment; filename=\""+rec.SafeName()+"\"")
}
http.ServeContent(w, r, rec.SafeName(), rec.CreatedAt, f)
case http.MethodDelete:
if err := a.store.Delete(r.Context(), id); err != nil {
http.Error(w, err.Error(), 404)
return
}
w.WriteHeader(204)
default:
w.WriteHeader(http.StatusMethodNotAllowed)
}
}
// /v1/uploads (POST) and /v1/uploads/{uid}/ ...
func (a *App) uploadsRoot(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
w.WriteHeader(http.StatusMethodNotAllowed)
return
}
name := r.URL.Query().Get("name")
meta := r.URL.Query().Get("meta")
u, err := a.store.UploadInit(r.Context(), name, meta)
if err != nil {
http.Error(w, err.Error(), 500)
return
}
a.writeJSON(w, 201, u)
}
func (a *App) uploadsByID(w http.ResponseWriter, r *http.Request) {
rest := strings.TrimPrefix(r.URL.Path, "/v1/uploads/")
parts := strings.Split(rest, "/")
if len(parts) < 1 || parts[0] == "" {
http.NotFound(w, r)
return
}
uid := parts[0]
if len(parts) == 3 && parts[1] == "parts" {
n := atoiDefault(parts[2], -1)
if r.Method != http.MethodPut || n < 1 {
http.Error(w, "invalid part", 400)
return
}
if err := a.store.UploadPart(r.Context(), uid, n, r.Body); err != nil {
http.Error(w, err.Error(), 400)
return
}
w.WriteHeader(204)
return
}
if len(parts) == 2 && parts[1] == "complete" {
if r.Method != http.MethodPost {
w.WriteHeader(http.StatusMethodNotAllowed)
return
}
rec, err := a.store.UploadComplete(r.Context(), uid)
if err != nil {
http.Error(w, err.Error(), 400)
return
}
a.writeJSON(w, 201, rec)
return
}
if len(parts) == 1 && r.Method == http.MethodDelete {
if err := a.store.UploadAbort(r.Context(), uid); err != nil {
http.Error(w, err.Error(), 400)
return
}
w.WriteHeader(204)
return
}
http.NotFound(w, r)
}
func atoiDefault(s string, d int) int {
n, err := strconv.Atoi(s)
if err != nil {
return d
}
return n
}

132
cmd/filesvc/ui/app.js
View File

@@ -1,132 +0,0 @@
(function() {
const $ = sel => document.querySelector(sel);
const $$ = sel => Array.from(document.querySelectorAll(sel));
const state = { offset: 0, limit: 20, total: null };
function loadCfg() {
try { return JSON.parse(localStorage.getItem('cfg')) || {}; } catch { return {}; }
}
function saveCfg(cfg) { localStorage.setItem('cfg', JSON.stringify(cfg)); }
const cfg = loadCfg();
$('#apiKey').value = cfg.apiKey || '';
$('#baseUrl').value = cfg.baseUrl || '';
$('#saveCfg').onclick = () => {
cfg.apiKey = $('#apiKey').value.trim();
cfg.baseUrl = $('#baseUrl').value.trim();
saveCfg(cfg);
refresh();
};
function api(path, opts = {}) {
const base = cfg.baseUrl || '';
opts.headers = Object.assign({ 'X-API-Key': cfg.apiKey || '' }, opts.headers || {});
return fetch(base + path, opts).then(r => {
if (!r.ok) throw new Error(`${r.status} ${r.statusText}`);
const ct = r.headers.get('content-type') || '';
if (ct.includes('application/json')) return r.json();
return r.text();
});
}
async function refresh() {
const q = encodeURIComponent($('#q').value || '');
try {
const data = await api(`/v1/files?limit=${state.limit}&offset=${state.offset}&q=${q}`);
renderTable(data.items || []);
const next = data.next || 0;
state.hasNext = next > 0;
state.nextOffset = next;
$('#pageInfo').textContent = `offset ${state.offset}`;
} catch (e) {
alert('List failed: ' + e.message);
}
}
function renderTable(items) {
const tbody = $('#files tbody');
tbody.innerHTML = '';
const tpl = $('#rowTpl').content;
for (const it of items) {
const tr = tpl.cloneNode(true);
tr.querySelector('.id').textContent = it.id;
tr.querySelector('.name').textContent = it.name;
tr.querySelector('.size').textContent = human(it.size);
tr.querySelector('.created').textContent = new Date(it.createdAt).toLocaleString();
const act = tr.querySelector('.actions');
const dl = btn('Download', async () => {
const base = cfg.baseUrl || '';
const url = `${base}/v1/files/${it.id}?download=1`;
const a = document.createElement('a');
a.href = url; a.download = '';
a.click();
});
const meta = btn('Meta', async () => showMeta(it.id));
const del = btn('Delete', async () => {
if (!confirm('Delete file?')) return;
try { await api(`/v1/files/${it.id}`, { method:'DELETE' }); refresh(); } catch(e){ alert('Delete failed: '+e.message); }
});
act.append(dl, meta, del);
tbody.appendChild(tr);
}
}
function btn(text, on) { const b = document.createElement('button'); b.textContent = text; b.onclick = on; return b; }
function human(n) { if (n < 1024) return n + ' B'; const u=['KB','MB','GB','TB']; let i=-1; do { n/=1024; i++; } while(n>=1024 && i<u.length-1); return n.toFixed(1)+' '+u[i]; }
$('#refresh').onclick = () => { state.offset = 0; refresh(); };
$('#q').addEventListener('keydown', e => { if (e.key==='Enter') { state.offset=0; refresh(); } });
$('#prev').onclick = () => { state.offset = Math.max(0, state.offset - state.limit); refresh(); };
$('#next').onclick = () => { if (state.hasNext) { state.offset = state.nextOffset; refresh(); } };
// Upload form
$('#uploadForm').addEventListener('submit', async (e) => {
e.preventDefault();
const f = $('#fileInput').files[0];
if (!f) return alert('Pick a file');
const meta = $('#metaInput').value.trim();
const fd = new FormData();
fd.append('file', f);
fd.append('meta', meta);
try { await api('/v1/files?meta='+encodeURIComponent(meta), { method: 'POST', body: fd }); refresh(); } catch(e){ alert('Upload failed: '+e.message); }
});
// Chunked upload
$('#chunkInit').onclick = async () => {
try {
const name = $('#chunkName').value.trim() || 'file';
const meta = $('#chunkMeta').value.trim();
const r = await api(`/v1/uploads?name=${encodeURIComponent(name)}&meta=${encodeURIComponent(meta)}`, { method:'POST' });
$('#chunkId').textContent = r.id;
} catch(e){ alert('Init failed: '+e.message); }
};
$('#chunkPut').onclick = async () => {
const uid = $('#chunkId').textContent.trim();
const part = parseInt($('#chunkPart').value,10) || 1;
const file = $('#chunkFile').files[0];
if (!uid) return alert('Init first');
if (!file) return alert('Choose a file (this will send the whole file as one part).');
try { await api(`/v1/uploads/${uid}/parts/${part}`, { method:'PUT', body: file }); alert('Part uploaded'); } catch(e){ alert('PUT failed: '+e.message); }
};
$('#chunkComplete').onclick = async () => {
const uid = $('#chunkId').textContent.trim(); if (!uid) return;
try { await api(`/v1/uploads/${uid}/complete`, { method:'POST' }); refresh(); } catch(e){ alert('Complete failed: '+e.message); }
};
$('#chunkAbort').onclick = async () => {
const uid = $('#chunkId').textContent.trim(); if (!uid) return;
try { await api(`/v1/uploads/${uid}`, { method:'DELETE' }); $('#chunkId').textContent=''; alert('Aborted'); } catch(e){ alert('Abort failed: '+e.message); }
};
async function showMeta(id) {
try {
const rec = await api(`/v1/files/${id}/meta`);
const json = prompt('Edit meta as JSON (object of string:string)', JSON.stringify(rec.meta||{}));
if (json == null) return;
const obj = JSON.parse(json);
await api(`/v1/files/${id}/meta`, { method:'PUT', headers:{'Content-Type':'application/json'}, body: JSON.stringify(obj) });
refresh();
} catch(e){ alert('Meta failed: '+e.message); }
}
refresh();
})();

77
cmd/filesvc/ui/index.html
View File

@@ -1,77 +0,0 @@
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8" />
<meta name="viewport" content="width=device-width, initial-scale=1" />
<title>File Service UI</title>
<link rel="stylesheet" href="/static/ui/style.css" />
</head>
<body>
<header>
<h1>File Service</h1>
<div class="cfg">
<label>API Base <input id="baseUrl" value="" placeholder="(same origin)"/></label>
<label>API Key <input id="apiKey" placeholder="X-API-Key"/></label>
<button id="saveCfg">Save</button>
</div>
</header>
<main>
<section class="card">
<h2>Upload</h2>
<form id="uploadForm">
<input type="file" id="fileInput" name="file" required />
<input type="text" id="metaInput" placeholder="meta e.g. project=alpha,owner=alice" />
<button type="submit">Upload</button>
</form>
<details>
<summary>Chunked upload</summary>
<div class="chunk">
<input type="text" id="chunkName" placeholder="filename"/>
<input type="text" id="chunkMeta" placeholder="meta key=val,..."/>
<button id="chunkInit">Init</button>
<span id="chunkId"></span>
<div>
<input type="file" id="chunkFile"/>
<input type="number" id="chunkPart" min="1" value="1"/>
<button id="chunkPut">PUT Part</button>
<button id="chunkComplete">Complete</button>
<button id="chunkAbort">Abort</button>
</div>
</div>
</details>
</section>
<section class="card">
<h2>Files</h2>
<div class="toolbar">
<input type="search" id="q" placeholder="search by name"/>
<button id="refresh">Refresh</button>
</div>
<table id="files">
<thead>
<tr><th>ID</th><th>Name</th><th>Size</th><th>Created</th><th>Actions</th></tr>
</thead>
<tbody></tbody>
</table>
<div class="pager">
<button id="prev">Prev</button>
<span id="pageInfo"></span>
<button id="next">Next</button>
</div>
</section>
</main>
<template id="rowTpl">
<tr>
<td class="mono id"></td>
<td class="name"></td>
<td class="size"></td>
<td class="created"></td>
<td class="actions"></td>
</tr>
</template>
<script src="/static/ui/app.js"></script>
</body>
</html>

19
cmd/filesvc/ui/style.css
View File

@@ -1,19 +0,0 @@
:root { --bg: #0b0f14; --fg: #e6eef8; --muted: #9bb0c8; --card: #121923; --accent: #5aa9ff; }
* { box-sizing: border-box; }
body { margin: 0; font-family: ui-sans-serif, system-ui, -apple-system, Segoe UI, Roboto, sans-serif; background: var(--bg); color: var(--fg); }
header { display: flex; justify-content: space-between; align-items: center; padding: 16px 20px; background: #0e141b; border-bottom: 1px solid #2a3543; }
h1 { margin: 0; font-size: 20px; }
.cfg label { margin-right: 8px; font-size: 12px; color: var(--muted); }
.cfg input { margin-left: 6px; padding: 6px 8px; background: #0c1219; border: 1px solid #2a3543; color: var(--fg); border-radius: 6px; }
button { padding: 8px 12px; border: 1px solid #2a3543; background: #111a24; color: var(--fg); border-radius: 8px; cursor: pointer; }
button:hover { border-color: var(--accent); }
main { padding: 20px; max-width: 1100px; margin: 0 auto; }
.card { background: var(--card); border: 1px solid #1f2a38; border-radius: 14px; padding: 16px; margin-bottom: 16px; box-shadow: 0 6px 20px rgba(0,0,0,.25); }
.toolbar { display:flex; gap: 8px; align-items: center; margin-bottom: 10px; }
table { width: 100%; border-collapse: collapse; }
th, td { text-align: left; padding: 8px; border-bottom: 1px solid #213043; }
.mono { font-family: ui-monospace, SFMono-Regular, Menlo, Consolas, monospace; font-size: 12px; }
.name { max-width: 340px; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; }
.pager { display:flex; gap: 8px; align-items:center; justify-content:flex-end; padding-top: 8px; }
.actions button { margin-right: 6px; }
summary { cursor: pointer; }

398
cmd/unified/main.go Normal file
View File

@@ -0,0 +1,398 @@
package main
import (
"context"
"encoding/json"
"errors"
"fmt"
"log"
"net/http"
"os"
"os/signal"
"strconv"
"strings"
"syscall"
"time"
// ← Passe diese Import-Pfade an dein go.mod an
"git.send.nrw/sendnrw/decent-webui/internal/admin"
"git.send.nrw/sendnrw/decent-webui/internal/filesvc"
"git.send.nrw/sendnrw/decent-webui/internal/mesh"
)
/*** Config ***/
func loadConfig() AppConfig {
// HTTP
httpAddr := getenvDefault("ADDR", ":8080")
// API
apiKey := os.Getenv("FILE_SERVICE_API_KEY")
// Admin UI (BasicAuth optional)
adminUser := os.Getenv("ADMIN_USER")
adminPass := os.Getenv("ADMIN_PASS")
// Mesh (mit sinnvollen Defaults)
m := mesh.Config{
BindAddr: getenvDefault("MESH_BIND", ":9090"),
AdvertURL: os.Getenv("MESH_ADVERT"), // kann leer sein → wir leiten ab
Seeds: splitCSV(os.Getenv("MESH_SEEDS")),
ClusterSecret: os.Getenv("MESH_CLUSTER_SECRET"),
EnableDiscovery: parseBoolEnv("MESH_ENABLE_DISCOVERY", false),
DiscoveryAddress: getenvDefault("MESH_DISCOVERY_ADDR", "239.8.8.8:9898"),
}
// Wenn keine AdvertURL gesetzt ist, versuche eine sinnvolle Herleitung:
if strings.TrimSpace(m.AdvertURL) == "" {
m.AdvertURL = inferAdvertURL(m.BindAddr)
log.Printf("[mesh] MESH_ADVERT nicht gesetzt – abgeleitet: %s", m.AdvertURL)
}
// Minimal-Validierung mit hilfreicher Meldung
if strings.TrimSpace(m.BindAddr) == "" {
log.Fatal("MESH_BIND fehlt (z.B. :9090)")
}
if strings.TrimSpace(m.AdvertURL) == "" {
log.Fatal("MESH_ADVERT fehlt und konnte nicht abgeleitet werden (z.B. http://unified_a:9090)")
}
if strings.TrimSpace(m.ClusterSecret) == "" {
log.Printf("[mesh] WARN: MESH_CLUSTER_SECRET ist leer – für produktive Netze unbedingt setzen!")
}
return AppConfig{
HTTPAddr: httpAddr,
APIKey: apiKey,
AdminUser: adminUser,
AdminPass: adminPass,
Mesh: m,
}
}
// --- Helpers
func getenvDefault(k, def string) string {
v := os.Getenv(k)
if v == "" {
return def
}
return v
}
func parseBoolEnv(k string, def bool) bool {
v := strings.ToLower(strings.TrimSpace(os.Getenv(k)))
if v == "" {
return def
}
return v == "1" || v == "true" || v == "yes" || v == "on"
}
func splitCSV(s string) []string {
s = strings.TrimSpace(s)
if s == "" {
return nil
}
parts := strings.Split(s, ",")
out := make([]string, 0, len(parts))
seen := map[string]struct{}{}
for _, p := range parts {
p = strings.TrimSpace(p)
if p == "" {
continue
}
if _, ok := seen[p]; ok {
continue
}
seen[p] = struct{}{}
out = append(out, p)
}
return out
}
// inferAdvertURL baut eine brauchbare Default-AdvertURL:
// - MESH_ADVERT_HOST (wenn gesetzt) → z.B. Service-Name aus Compose
// - sonst HOSTNAME
// - sonst "localhost"
//
// Port wird aus MESH_BIND entnommen (z.B. ":9090" → 9090)
func inferAdvertURL(meshBind string) string {
host := strings.TrimSpace(os.Getenv("MESH_ADVERT_HOST"))
if host == "" {
host = strings.TrimSpace(os.Getenv("HOSTNAME"))
}
if host == "" {
host = "localhost"
}
port := "9090"
if i := strings.LastIndex(meshBind, ":"); i != -1 && len(meshBind) > i+1 {
port = meshBind[i+1:]
}
return fmt.Sprintf("http://%s:%s", host, port)
}
type AppConfig struct {
HTTPAddr string
APIKey string
AdminUser string
AdminPass string
Mesh mesh.Config
}
/*** Middleware ***/
func authMiddleware(apiKey string, next http.Handler) http.Handler {
// Dev-Mode: ohne API-Key kein Auth-Zwang
if strings.TrimSpace(apiKey) == "" {
return next
}
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
got := r.Header.Get("Authorization")
if !strings.HasPrefix(got, "Bearer ") || strings.TrimPrefix(got, "Bearer ") != apiKey {
http.Error(w, "unauthorized", http.StatusUnauthorized)
return
}
next.ServeHTTP(w, r)
})
}
func cors(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
// Passe nach Bedarf an (Origin-Whitelist etc.)
w.Header().Set("Access-Control-Allow-Origin", "*")
w.Header().Set("Access-Control-Allow-Headers", "Authorization, Content-Type")
w.Header().Set("Access-Control-Allow-Methods", "GET,POST,PUT,DELETE,OPTIONS")
if r.Method == http.MethodOptions {
w.WriteHeader(http.StatusNoContent)
return
}
next.ServeHTTP(w, r)
})
}
func accessLog(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
start := time.Now()
next.ServeHTTP(w, r)
log.Printf("%s %s %s", r.Method, r.URL.Path, time.Since(start))
})
}
/*** HTTP helpers ***/
func writeJSON(w http.ResponseWriter, code int, v any) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(code)
_ = json.NewEncoder(w).Encode(v)
}
/*** API-Routen ***/
func fileRoutes(mux *http.ServeMux, store filesvc.MeshStore) {
// Health
mux.HandleFunc("/api/v1/health", func(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
})
// List + Create
mux.HandleFunc("/api/v1/items", func(w http.ResponseWriter, r *http.Request) {
switch r.Method {
case http.MethodGet:
nextStr := r.URL.Query().Get("next")
var next filesvc.ID
if nextStr != "" {
if n, err := strconv.ParseInt(nextStr, 10, 64); err == nil {
next = filesvc.ID(n)
}
}
items, nextOut, err := store.List(r.Context(), next, 100)
if err != nil {
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()})
return
}
writeJSON(w, http.StatusOK, map[string]any{"items": items, "next": nextOut})
case http.MethodPost:
var in struct {
Name string `json:"name"`
}
if err := json.NewDecoder(r.Body).Decode(&in); err != nil {
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid json"})
return
}
it, err := store.Create(r.Context(), in.Name)
if err != nil {
writeJSON(w, http.StatusBadRequest, map[string]string{"error": err.Error()})
return
}
writeJSON(w, http.StatusCreated, it)
default:
w.WriteHeader(http.StatusMethodNotAllowed)
}
})
// Rename
mux.HandleFunc("/api/v1/items/rename", func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
w.WriteHeader(http.StatusMethodNotAllowed)
return
}
var in struct {
ID filesvc.ID `json:"id"`
Name string `json:"name"`
}
if err := json.NewDecoder(r.Body).Decode(&in); err != nil {
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid json"})
return
}
it, err := store.Rename(r.Context(), in.ID, in.Name)
if err != nil {
status := http.StatusBadRequest
if errors.Is(err, filesvc.ErrNotFound) {
status = http.StatusNotFound
}
writeJSON(w, status, map[string]string{"error": err.Error()})
return
}
writeJSON(w, http.StatusOK, it)
})
// Delete
mux.HandleFunc("/api/v1/items/delete", func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
w.WriteHeader(http.StatusMethodNotAllowed)
return
}
var in struct {
ID filesvc.ID `json:"id"`
}
if err := json.NewDecoder(r.Body).Decode(&in); err != nil {
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid json"})
return
}
it, err := store.Delete(r.Context(), in.ID)
if err != nil {
status := http.StatusBadRequest
if errors.Is(err, filesvc.ErrNotFound) {
status = http.StatusNotFound
}
writeJSON(w, status, map[string]string{"error": err.Error()})
return
}
writeJSON(w, http.StatusOK, it)
})
}
/*** Mesh <-> Store Mapping (falls Typen getrennt sind) ***/
func toMeshSnapshot(s filesvc.Snapshot) mesh.Snapshot {
out := mesh.Snapshot{Items: make([]mesh.Item, 0, len(s.Items))}
for _, it := range s.Items {
out.Items = append(out.Items, mesh.Item{
ID: int64(it.ID),
Name: it.Name,
UpdatedAt: it.UpdatedAt,
Deleted: it.Deleted,
})
}
return out
}
func fromMeshSnapshot(ms mesh.Snapshot) filesvc.Snapshot {
out := filesvc.Snapshot{Items: make([]filesvc.File, 0, len(ms.Items))}
for _, it := range ms.Items {
out.Items = append(out.Items, filesvc.File{
ID: filesvc.ID(it.ID),
Name: it.Name,
UpdatedAt: it.UpdatedAt,
Deleted: it.Deleted,
})
}
return out
}
/*** main ***/
func main() {
cfg := loadConfig()
// Domain-Store (mesh-fähig)
st := filesvc.NewMemStore()
// Mesh starten
mcfg := mesh.FromEnv()
mnode, err := mesh.New(mcfg, mesh.Callbacks{
GetSnapshot: func(ctx context.Context) (mesh.Snapshot, error) {
s, err := st.Snapshot(ctx)
if err != nil {
return mesh.Snapshot{}, err
}
return toMeshSnapshot(s), nil
},
ApplyRemote: func(ctx context.Context, s mesh.Snapshot) error {
return st.ApplyRemote(ctx, fromMeshSnapshot(s))
},
})
if err != nil {
log.Fatalf("mesh init: %v", err)
}
go func() {
log.Printf("[mesh] listening on %s advertise %s seeds=%v discovery=%v",
mcfg.BindAddr, mcfg.AdvertURL, mcfg.Seeds, mcfg.EnableDiscovery)
if err := mnode.Serve(); err != nil && !errors.Is(err, http.ErrServerClosed) {
log.Fatalf("mesh serve: %v", err)
}
}()
// Root-Mux
root := http.NewServeMux()
// API (Bearer-Auth)
apiMux := http.NewServeMux()
fileRoutes(apiMux, st)
root.Handle("/api/", authMiddleware(cfg.APIKey, apiMux))
// Admin-UI (optional BasicAuth via ADMIN_USER/ADMIN_PASS)
adminRoot := http.NewServeMux()
admin.Register(adminRoot, admin.Deps{Store: st, Mesh: mnode})
adminUser := os.Getenv("ADMIN_USER")
adminPass := os.Getenv("ADMIN_PASS")
if strings.TrimSpace(adminUser) != "" {
wrapped := admin.BasicAuth(adminUser, adminPass, adminRoot)
root.Handle("/admin", wrapped)
root.Handle("/admin/", wrapped)
} else {
root.Handle("/admin", adminRoot)
root.Handle("/admin/", adminRoot)
}
// Startseite → /admin
root.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
http.Redirect(w, r, "/admin", http.StatusFound)
})
// Finaler Handler-Stack
handler := cors(accessLog(root))
srv := &http.Server{
Addr: cfg.HTTPAddr,
Handler: handler,
ReadHeaderTimeout: 5 * time.Second,
}
// Graceful shutdown
go func() {
log.Printf("http listening on %s (api=/api/v1, admin=/admin)", cfg.Mesh.BindAddr)
if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
log.Fatalf("http server: %v", err)
}
}()
// OS-Signale abfangen
stop := make(chan os.Signal, 1)
signal.Notify(stop, syscall.SIGINT, syscall.SIGTERM)
<-stop
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
_ = srv.Shutdown(ctx)
_ = mnode.Close(ctx)
log.Println("shutdown complete")
}

146
internal/admin/admin.go Normal file
View File

@@ -0,0 +1,146 @@
package admin
import (
"context"
_ "embed"
"html/template"
"net/http"
"strconv"
"strings"
"time"
"git.send.nrw/sendnrw/decent-webui/internal/filesvc"
"git.send.nrw/sendnrw/decent-webui/internal/mesh"
)
/*** Templates einbetten ***/
//go:embed tpl/layout.html
var layoutHTML string
//go:embed tpl/partials_items.html
var itemsPartialHTML string
//go:embed tpl/partials_peers.html
var peersPartialHTML string
var (
tplLayout = template.Must(template.New("layout").Parse(layoutHTML))
tplItems = template.Must(template.New("items").Funcs(template.FuncMap{
"timeRFC3339": func(unixNano int64) string {
if unixNano == 0 {
return ""
}
return time.Unix(0, unixNano).UTC().Format(time.RFC3339)
},
}).Parse(itemsPartialHTML))
tplPeers = template.Must(template.New("peers").Parse(peersPartialHTML))
)
type Deps struct {
Store filesvc.MeshStore
Mesh *mesh.Node
}
// Register hängt alle /admin Routen ein.
// Auth liegt optional VOR Register (BasicAuth-Middleware), siehe main.go.
func Register(mux *http.ServeMux, d Deps) {
// Dashboard
mux.HandleFunc("/admin", func(w http.ResponseWriter, r *http.Request) {
renderLayout(w, r, "Files", "/admin/items")
})
// Partials
mux.HandleFunc("/admin/items", func(w http.ResponseWriter, r *http.Request) {
// Liste rendern (Pagination optional via ?next=)
nextQ := strings.TrimSpace(r.URL.Query().Get("next"))
var nextID filesvc.ID
if nextQ != "" {
if n, err := strconv.ParseInt(nextQ, 10, 64); err == nil {
nextID = filesvc.ID(n)
}
}
items, nextOut, _ := d.Store.List(r.Context(), nextID, 100)
_ = tplItems.Execute(w, map[string]any{
"Items": items,
"Next": nextOut,
})
})
mux.HandleFunc("/admin/peers", func(w http.ResponseWriter, r *http.Request) {
peers := d.Mesh.PeerList()
_ = tplPeers.Execute(w, map[string]any{
"Peers": peers,
"Now": time.Now(),
})
})
// Actions (HTMX POSTs)
mux.HandleFunc("/admin/items/create", func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}
name := strings.TrimSpace(r.FormValue("name"))
if name != "" {
_, _ = d.Store.Create(r.Context(), name)
_ = d.Mesh.SyncNow(r.Context()) // prompt push (best effort)
}
// Nach Aktion Items partial zurückgeben (HTMX swap)
http.Redirect(w, r, "/admin/items", http.StatusSeeOther)
})
mux.HandleFunc("/admin/items/rename", func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}
idStr := r.FormValue("id")
newName := strings.TrimSpace(r.FormValue("name"))
if id, err := strconv.ParseInt(idStr, 10, 64); err == nil && newName != "" {
_, _ = d.Store.Rename(r.Context(), filesvc.ID(id), newName)
_ = d.Mesh.SyncNow(r.Context())
}
http.Redirect(w, r, "/admin/items", http.StatusSeeOther)
})
mux.HandleFunc("/admin/items/delete", func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}
if id, err := strconv.ParseInt(r.FormValue("id"), 10, 64); err == nil {
_, _ = d.Store.Delete(r.Context(), filesvc.ID(id))
_ = d.Mesh.SyncNow(r.Context())
}
http.Redirect(w, r, "/admin/items", http.StatusSeeOther)
})
mux.HandleFunc("/admin/mesh/syncnow", func(w http.ResponseWriter, r *http.Request) {
_ = d.Mesh.SyncNow(context.Background())
http.Redirect(w, r, "/admin/peers", http.StatusSeeOther)
})
}
func renderLayout(w http.ResponseWriter, _ *http.Request, active string, initial string) {
_ = tplLayout.Execute(w, map[string]any{
"Active": active,
"Init": initial, // initialer HTMX Swap-Endpunkt
})
}
/*** Optional: einfache BasicAuth (siehe main.go) ***/
func BasicAuth(user, pass string, next http.Handler) http.Handler {
if strings.TrimSpace(user) == "" {
return next // deaktiviert
}
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
u, p, ok := r.BasicAuth()
if !ok || u != user || p != pass {
w.Header().Set("WWW-Authenticate", `Basic realm="admin"`)
http.Error(w, "unauthorized", http.StatusUnauthorized)
return
}
next.ServeHTTP(w, r)
})
}

39
internal/admin/tpl/layout.html Normal file
View File

@@ -0,0 +1,39 @@
<!doctype html>
<html lang="de">
<head>
<meta charset="utf-8"/>
<meta name="viewport" content="width=device-width,initial-scale=1"/>
<title>Admin</title>
<script src="https://unpkg.com/htmx.org@1.9.12"></script>
<style>
:root { --bg:#0b1220; --card:#121a2b; --muted:#94a3b8; --text:#e5e7eb; --accent:#4f46e5; }
html,body { margin:0; background:var(--bg); color:var(--text); font-family: ui-sans-serif, system-ui, -apple-system, Segoe UI, Roboto, Ubuntu; }
.wrap { max-width: 980px; margin: 40px auto; padding: 0 16px; }
.nav { display:flex; gap:12px; margin-bottom:16px; }
.btn { background:var(--card); border:1px solid #243044; padding:8px 12px; border-radius:10px; color:var(--text); cursor:pointer; }
.btn:hover { border-color:#3b4a66; }
.btn-primary { background: var(--accent); border-color: var(--accent); color:white; }
.card { background:var(--card); border:1px solid #1f2937; border-radius:16px; padding:16px; box-shadow: 0 6px 30px rgba(0,0,0,.25); }
.row { display:flex; gap:16px; flex-wrap:wrap; }
table { width:100%; border-collapse: collapse; }
th, td { border-bottom: 1px solid #1f2937; padding:10px; text-align:left; }
input[type="text"] { background:#0f1626; border:1px solid #263246; color:var(--text); padding:8px 10px; border-radius:10px; width:100%; }
small { color: var(--muted); }
.muted { color: var(--muted); }
.pill { font-size: 12px; padding: 2px 8px; border:1px solid #2b364b; border-radius:999px; background:#0f1626; color:#bcd; }
</style>
</head>
<body>
<div class="wrap">
<h1 style="margin:0 0 10px 0;">Unified Admin</h1>
<div class="nav">
<button class="btn" hx-get="/admin/items" hx-target="#main" hx-swap="innerHTML">Dateien</button>
<button class="btn" hx-get="/admin/peers" hx-target="#main" hx-swap="innerHTML">Mesh</button>
</div>
<div id="main" class="card" hx-get="{{.Init}}" hx-trigger="load" hx-swap="innerHTML">
<div class="muted">Lade…</div>
</div>
<div style="margin-top:14px"><small>© Admin UI</small></div>
</div>
</body>
</html>

57
internal/admin/tpl/partials_items.html Normal file
View File

@@ -0,0 +1,57 @@
<div class="row">
<div style="flex: 1 1 360px">
<form hx-post="/admin/items/create" hx-target="#items" hx-get="/admin/items" hx-swap="outerHTML">
<label>Neue Datei</label>
<div style="display:flex; gap:8px; margin-top:6px">
<input type="text" name="name" placeholder="z.B. notes.txt" required>
<button class="btn btn-primary" type="submit">Anlegen</button>
</div>
</form>
</div>
</div>
<div id="items" style="margin-top:14px">
<table>
<thead>
<tr>
<th>ID</th><th>Name</th><th>Updated</th><th style="width:200px">Aktionen</th>
</tr>
</thead>
<tbody>
{{ range .Items }}
<tr>
<td>{{ .ID }}</td>
<td>{{ .Name }}</td>
<td><small class="muted">{{ printf "%.19s" (timeRFC3339 .UpdatedAt) }}</small></td>
<td>
<form style="display:inline-flex; gap:6px"
hx-post="/admin/items/rename"
hx-target="#items" hx-get="/admin/items" hx-swap="outerHTML">
<input type="hidden" name="id" value="{{ .ID }}">
<input type="text" name="name" placeholder="Neuer Name">
<button class="btn" type="submit" title="Umbenennen">Rename</button>
</form>
<form style="display:inline"
hx-post="/admin/items/delete"
hx-target="#items" hx-get="/admin/items" hx-swap="outerHTML"
onsubmit="return confirm('Wirklich löschen?');">
<input type="hidden" name="id" value="{{ .ID }}">
<button class="btn" type="submit" title="Löschen">Delete</button>
</form>
</td>
</tr>
{{ else }}
<tr><td colspan="4" class="muted">Keine Dateien vorhanden.</td></tr>
{{ end }}
</tbody>
</table>
{{ if .Next }}
<div style="margin-top:10px">
<button class="btn"
hx-get="/admin/items?next={{ .Next }}"
hx-target="#items" hx-swap="outerHTML">Mehr laden</button>
<span class="pill">next={{ .Next }}</span>
</div>
{{ end }}
</div>

23
internal/admin/tpl/partials_peers.html Normal file
View File

@@ -0,0 +1,23 @@
<div style="display:flex; justify-content:space-between; align-items:center">
<h3 style="margin:0">Mesh Peers</h3>
<form hx-post="/admin/mesh/syncnow" hx-target="#peers" hx-get="/admin/peers" hx-swap="outerHTML">
<button class="btn btn-primary" type="submit">Jetzt synchronisieren</button>
</form>
</div>
<div id="peers" style="margin-top:10px">
<table>
<thead><tr><th>URL</th><th>Self</th><th>Last Seen</th></tr></thead>
<tbody>
{{ range .Peers }}
<tr>
<td>{{ .URL }}</td>
<td>{{ if .Self }}✅{{ else }}—{{ end }}</td>
<td><small class="muted">{{ .LastSeen }}</small></td>
</tr>
{{ else }}
<tr><td colspan="3" class="muted">Keine Peers bekannt.</td></tr>
{{ end }}
</tbody>
</table>
<div class="muted" style="margin-top:8px"><small>Stand: {{ .Now }}</small></div>
</div>

195
internal/filesvc/memstore.go Normal file
View File

@@ -0,0 +1,195 @@
package filesvc
import (
"context"
"slices"
"strings"
"sync"
"time"
)
type MemStore struct {
mu sync.Mutex
items map[ID]File
next ID
// optionales Eventing
subs []chan ChangeEvent
}
func NewMemStore() *MemStore {
return &MemStore{
items: make(map[ID]File),
next: 1,
}
}
/*** Store ***/
func (m *MemStore) Get(_ context.Context, id ID) (File, error) {
m.mu.Lock()
defer m.mu.Unlock()
it, ok := m.items[id]
if !ok || it.Deleted {
return File{}, ErrNotFound
}
return it, nil
}
func (m *MemStore) List(_ context.Context, next ID, limit int) ([]File, ID, error) {
m.mu.Lock()
defer m.mu.Unlock()
if limit <= 0 || limit > 1000 {
limit = 100
}
ids := make([]ID, 0, len(m.items))
for id := range m.items {
ids = append(ids, id)
}
slices.Sort(ids)
start := 0
if next > 0 {
for i, id := range ids {
if id >= next {
start = i
break
}
}
}
end := start + limit
if end > len(ids) {
end = len(ids)
}
out := make([]File, 0, end-start)
for _, id := range ids[start:end] {
it := m.items[id]
if !it.Deleted {
out = append(out, it)
}
}
var nextOut ID
if end < len(ids) {
nextOut = ids[end]
}
return out, nextOut, nil
}
func (m *MemStore) Create(_ context.Context, name string) (File, error) {
name = strings.TrimSpace(name)
if name == "" {
return File{}, ErrBadInput
}
m.mu.Lock()
defer m.mu.Unlock()
now := time.Now().UnixNano()
it := File{ID: m.next, Name: name, UpdatedAt: now}
m.items[it.ID] = it
m.next++
m.emit(it)
return it, nil
}
func (m *MemStore) Rename(_ context.Context, id ID, newName string) (File, error) {
newName = strings.TrimSpace(newName)
if newName == "" {
return File{}, ErrBadInput
}
m.mu.Lock()
defer m.mu.Unlock()
it, ok := m.items[id]
if !ok || it.Deleted {
return File{}, ErrNotFound
}
it.Name = newName
it.UpdatedAt = time.Now().UnixNano()
m.items[id] = it
m.emit(it)
return it, nil
}
func (m *MemStore) Delete(_ context.Context, id ID) (File, error) {
m.mu.Lock()
defer m.mu.Unlock()
it, ok := m.items[id]
if !ok {
return File{}, ErrNotFound
}
if it.Deleted {
return it, nil
}
it.Deleted = true
it.UpdatedAt = time.Now().UnixNano()
m.items[id] = it
m.emit(it)
return it, nil
}
/*** Replicable ***/
func (m *MemStore) Snapshot(_ context.Context) (Snapshot, error) {
m.mu.Lock()
defer m.mu.Unlock()
s := Snapshot{Items: make([]File, 0, len(m.items))}
for _, it := range m.items {
s.Items = append(s.Items, it) // inkl. Tombstones
}
return s, nil
}
func (m *MemStore) ApplyRemote(_ context.Context, s Snapshot) error {
m.mu.Lock()
defer m.mu.Unlock()
for _, ri := range s.Items {
li, ok := m.items[ri.ID]
if !ok || ri.UpdatedAt > li.UpdatedAt {
m.items[ri.ID] = ri
if ri.ID >= m.next {
m.next = ri.ID + 1
}
m.emitLocked(ri)
}
}
return nil
}
/*** Watchable (optional) ***/
func (m *MemStore) Watch(stop <-chan struct{}) <-chan ChangeEvent {
ch := make(chan ChangeEvent, 32)
m.mu.Lock()
m.subs = append(m.subs, ch)
m.mu.Unlock()
go func() {
<-stop
m.mu.Lock()
// entferne ch aus subs
for i, s := range m.subs {
if s == ch {
m.subs = append(m.subs[:i], m.subs[i+1:]...)
break
}
}
m.mu.Unlock()
close(ch)
}()
return ch
}
func (m *MemStore) emit(it File) {
m.emitLocked(it) // mu wird im Aufrufer gehalten
}
func (m *MemStore) emitLocked(it File) {
ev := ChangeEvent{At: time.Now(), Item: it}
for _, s := range m.subs {
select {
case s <- ev:
default: /* drop wenn voll */
}
}
}

78
internal/filesvc/store.go Normal file
View File

@@ -0,0 +1,78 @@
package filesvc
import (
"context"
"errors"
"time"
)
/*** Domain ***/
type ID = int64
type File struct {
ID ID `json:"id"`
Name string `json:"name"`
// weitere Metadaten optional: Size, Hash, Owner, Tags, ...
UpdatedAt int64 `json:"updatedAt"` // UnixNano für LWW
Deleted bool `json:"deleted"` // Tombstone für Mesh-Delete
}
/*** Fehler ***/
var (
ErrNotFound = errors.New("file not found")
ErrBadInput = errors.New("bad input")
ErrConflict = errors.New("conflict")
ErrForbidden = errors.New("forbidden")
ErrTransient = errors.New("transient")
)
/*** Basis-API (lokal nutzbar) ***/
type Store interface {
// Lesen & Auflisten
Get(ctx context.Context, id ID) (File, error)
List(ctx context.Context, next ID, limit int) (items []File, nextOut ID, err error)
// Mutationen mit LWW-Semantik (UpdatedAt wird intern gesetzt, außer bei ApplyRemote)
Create(ctx context.Context, name string) (File, error)
Rename(ctx context.Context, id ID, newName string) (File, error)
Delete(ctx context.Context, id ID) (File, error)
}
/*** Mesh-Replikation ***/
type Snapshot struct {
Items []File `json:"items"`
}
type Replicable interface {
// Snapshot liefert den vollständigen aktuellen Stand (inkl. Tombstones).
Snapshot(ctx context.Context) (Snapshot, error)
// ApplyRemote wendet LWW an. next-ID wird dabei korrekt fortgeschrieben.
ApplyRemote(ctx context.Context, s Snapshot) error
}
/*** Events (optional) ***/
// ChangeEvent kann genutzt werden, um proaktive Mesh-Pushes zu triggern.
// Bei deiner Pull-basierten Anti-Entropy ist es optional.
type ChangeEvent struct {
At time.Time
Item File
}
// Watch gibt Änderungen aus; close(stop) beendet den Stream.
// Eine Noop-Implementierung ist erlaubt, wenn Pull-Sync genügt.
type Watchable interface {
Watch(stop <-chan struct{}) <-chan ChangeEvent
}
/*** Kombiniertes Interface ***/
type MeshStore interface {
Store
Replicable
Watchable // optional – kann Noop sein
}

451
internal/mesh/mesh.go Normal file
View File

@@ -0,0 +1,451 @@
package mesh
import (
"bytes"
"context"
"crypto/hmac"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"hash/crc32"
"io"
"log"
"net"
"net/http"
"os"
"slices"
"strings"
"sync"
"time"
)
/*** Types & Config ***/
type Config struct {
BindAddr string // e.g. ":9090"
AdvertURL string // e.g. "http://10.0.0.5:9090"
Seeds []string // other peers' mesh base URLs
ClusterSecret string // HMAC key
EnableDiscovery bool
DiscoveryAddress string // "239.8.8.8:9898"
}
type Peer struct {
URL string `json:"url"`
LastSeen time.Time `json:"lastSeen"`
Self bool `json:"self"`
OwnerHint int `json:"ownerHint"` // optional
}
type Item struct {
ID int64 `json:"id"`
Name string `json:"name"`
UpdatedAt int64 `json:"updatedAt"`
Deleted bool `json:"deleted"` // <— NEU: Tombstone für Deletes
}
type Snapshot struct {
Items []Item `json:"items"`
}
// Callbacks that your app provides
type Callbacks struct {
GetSnapshot func(ctx context.Context) (Snapshot, error)
ApplyRemote func(ctx context.Context, s Snapshot) error
}
/*** Node ***/
type Node struct {
cfg Config
cbs Callbacks
self Peer
mu sync.RWMutex
peers map[string]*Peer
client *http.Client
srv *http.Server
stop chan struct{}
wg sync.WaitGroup
}
func New(cfg Config, cbs Callbacks) (*Node, error) {
if cfg.BindAddr == "" || cfg.AdvertURL == "" {
return nil, errors.New("mesh: BindAddr and AdvertURL required")
}
if cfg.ClusterSecret == "" {
return nil, errors.New("mesh: ClusterSecret required")
}
n := &Node{
cfg: cfg,
cbs: cbs,
self: Peer{URL: cfg.AdvertURL, LastSeen: time.Now(), Self: true},
peers: make(map[string]*Peer),
client: &http.Client{
Timeout: 5 * time.Second,
},
stop: make(chan struct{}),
}
return n, nil
}
/*** HMAC helpers ***/
func (n *Node) sign(b []byte) string {
m := hmac.New(sha256.New, []byte(n.cfg.ClusterSecret))
m.Write(b)
return hex.EncodeToString(m.Sum(nil))
}
func (n *Node) verify(b []byte, sig string) bool {
want := n.sign(b)
return hmac.Equal([]byte(want), []byte(sig))
}
/*** HTTP handlers (control plane) ***/
func (n *Node) helloHandler(w http.ResponseWriter, r *http.Request) {
body, _ := io.ReadAll(r.Body)
if !n.verify(body, r.Header.Get("X-Mesh-Sig")) {
http.Error(w, "bad signature", http.StatusUnauthorized)
return
}
var p Peer
if err := json.Unmarshal(body, &p); err != nil {
http.Error(w, "bad json", http.StatusBadRequest)
return
}
p.LastSeen = time.Now()
n.mu.Lock()
if existing, ok := n.peers[p.URL]; ok {
existing.LastSeen = p.LastSeen
} else if p.URL != n.self.URL {
cp := p
n.peers[p.URL] = &cp
}
n.mu.Unlock()
w.WriteHeader(http.StatusNoContent)
}
func (n *Node) peersHandler(w http.ResponseWriter, r *http.Request) {
n.mu.RLock()
defer n.mu.RUnlock()
var list []Peer
list = append(list, n.self)
for _, p := range n.peers {
list = append(list, *p)
}
writeJSON(w, http.StatusOK, list)
}
func (n *Node) syncHandler(w http.ResponseWriter, r *http.Request) {
// verify signature
body, _ := io.ReadAll(r.Body)
if !n.verify(body, r.Header.Get("X-Mesh-Sig")) {
http.Error(w, "bad signature", http.StatusUnauthorized)
return
}
var s Snapshot
if err := json.Unmarshal(body, &s); err != nil {
http.Error(w, "bad json", http.StatusBadRequest)
return
}
// apply
if err := n.cbs.ApplyRemote(r.Context(), s); err != nil {
http.Error(w, "apply error: "+err.Error(), http.StatusInternalServerError)
return
}
w.WriteHeader(http.StatusNoContent)
}
func writeJSON(w http.ResponseWriter, code int, v any) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(code)
_ = json.NewEncoder(w).Encode(v)
}
/*** Serve ***/
func (n *Node) Serve() error {
mux := http.NewServeMux()
mux.HandleFunc("/mesh/peers", n.peersHandler)
mux.HandleFunc("/mesh/hello", n.helloHandler)
mux.HandleFunc("/mesh/sync", n.syncHandler)
n.srv = &http.Server{Addr: n.cfg.BindAddr, Handler: mux}
// background loops
n.wg.Add(1)
go func() {
defer n.wg.Done()
n.loopSeeder()
}()
if n.cfg.EnableDiscovery && n.cfg.DiscoveryAddress != "" {
n.wg.Add(2)
go func() {
defer n.wg.Done()
n.loopBeaconSend()
}()
go func() {
defer n.wg.Done()
n.loopBeaconRecv()
}()
}
n.wg.Add(1)
go func() {
defer n.wg.Done()
n.loopAntiEntropy()
}()
// http server
errc := make(chan error, 1)
go func() {
errc <- n.srv.ListenAndServe()
}()
select {
case err := <-errc:
return err
case <-n.stop:
return http.ErrServerClosed
}
}
func (n *Node) Close(ctx context.Context) error {
close(n.stop)
if n.srv != nil {
_ = n.srv.Shutdown(ctx)
}
n.wg.Wait()
return nil
}
/*** Loops ***/
func (n *Node) loopSeeder() {
// attempt to hello known seeds every 5s at start, then every 30s
backoff := 5 * time.Second
for {
select {
case <-n.stop:
return
case <-time.After(backoff):
}
if len(n.cfg.Seeds) == 0 {
backoff = 30 * time.Second
continue
}
for _, s := range n.cfg.Seeds {
if s == "" || s == n.self.URL {
continue
}
_ = n.sendHello(s)
}
backoff = 30 * time.Second
}
}
func (n *Node) loopAntiEntropy() {
t := time.NewTicker(10 * time.Second)
defer t.Stop()
for {
select {
case <-n.stop:
return
case <-t.C:
n.mu.RLock()
targets := make([]string, 0, len(n.peers))
for url := range n.peers {
targets = append(targets, url)
}
n.mu.RUnlock()
if len(targets) == 0 {
continue
}
ctx, cancel := context.WithTimeout(context.Background(), 4*time.Second)
snap, err := n.cbs.GetSnapshot(ctx)
cancel()
if err != nil {
continue
}
for _, url := range targets {
_ = n.sendSync(url, snap)
}
}
}
}
func (n *Node) loopBeaconSend() {
addr, err := net.ResolveUDPAddr("udp", n.cfg.DiscoveryAddress)
if err != nil {
log.Printf("mesh beacon send resolve: %v", err)
return
}
conn, err := net.DialUDP("udp", nil, addr)
if err != nil {
log.Printf("mesh beacon send dial: %v", err)
return
}
defer conn.Close()
type beacon struct {
URL string `json:"url"`
}
t := time.NewTicker(5 * time.Second)
defer t.Stop()
for {
select {
case <-n.stop:
return
case <-t.C:
b, _ := json.Marshal(beacon{URL: n.self.URL})
_, _ = conn.Write(b)
}
}
}
func (n *Node) loopBeaconRecv() {
addr, err := net.ResolveUDPAddr("udp", n.cfg.DiscoveryAddress)
if err != nil {
log.Printf("mesh beacon recv resolve: %v", err)
return
}
// enable multicast receive
l, err := net.ListenMulticastUDP("udp", nil, addr)
if err != nil {
log.Printf("mesh beacon recv listen: %v", err)
return
}
defer l.Close()
_ = l.SetReadBuffer(1 << 16)
buf := make([]byte, 2048)
for {
select {
case <-n.stop:
return
default:
}
_ = l.SetDeadline(time.Now().Add(6 * time.Second))
nr, _, err := l.ReadFromUDP(buf)
if err != nil {
if ne, ok := err.(net.Error); ok && ne.Timeout() {
continue
}
continue
}
var msg struct{ URL string }
if err := json.Unmarshal(buf[:nr], &msg); err == nil {
if msg.URL != "" && msg.URL != n.self.URL {
_ = n.sendHello(msg.URL)
}
}
}
}
/*** Outgoing ***/
func (n *Node) sendHello(url string) error {
p := n.self
b, _ := json.Marshal(p)
req, _ := http.NewRequest(http.MethodPost, strings.TrimRight(url, "/")+"/mesh/hello", bytes.NewReader(b))
req.Header.Set("X-Mesh-Sig", n.sign(b))
resp, err := n.client.Do(req)
if err == nil {
io.Copy(io.Discard, resp.Body)
resp.Body.Close()
}
return err
}
func (n *Node) sendSync(url string, s Snapshot) error {
b, _ := json.Marshal(s)
req, _ := http.NewRequest(http.MethodPost, strings.TrimRight(url, "/")+"/mesh/sync", bytes.NewReader(b))
req.Header.Set("X-Mesh-Sig", n.sign(b))
resp, err := n.client.Do(req)
if err == nil {
io.Copy(io.Discard, resp.Body)
resp.Body.Close()
}
return err
}
// PeerList liefert eine Kopie der bekannten Peers inkl. Self.
func (n *Node) PeerList() []Peer {
n.mu.RLock()
defer n.mu.RUnlock()
out := make([]Peer, 0, len(n.peers)+1)
out = append(out, n.self)
for _, p := range n.peers {
cp := *p
out = append(out, cp)
}
return out
}
// SyncNow verschickt sofort den aktuellen Snapshot an alle bekannten Peers.
func (n *Node) SyncNow(ctx context.Context) error {
snap, err := n.cbs.GetSnapshot(ctx)
if err != nil {
return err
}
n.mu.RLock()
targets := make([]string, 0, len(n.peers))
for url := range n.peers {
targets = append(targets, url)
}
n.mu.RUnlock()
for _, u := range targets {
_ = n.sendSync(u, snap)
}
return nil
}
/*** Utilities ***/
// OwnerHint is a simple, optional mapping to distribute responsibility.
func OwnerHint(id int64, peers []string) int {
if len(peers) == 0 {
return 0
}
h := crc32.ChecksumIEEE([]byte(string(rune(id))))
return int(h % uint32(len(peers)))
}
// Helpers to load from ENV quickly
func FromEnv() Config {
return Config{
BindAddr: getenvDefault("MESH_BIND", ":9090"),
AdvertURL: os.Getenv("MESH_ADVERT"),
Seeds: splitCSV(os.Getenv("MESH_SEEDS")),
ClusterSecret: os.Getenv("MESH_CLUSTER_SECRET"),
EnableDiscovery: os.Getenv("MESH_ENABLE_DISCOVERY") == "true",
DiscoveryAddress: getenvDefault("MESH_DISCOVERY_ADDR", "239.8.8.8:9898"),
}
}
func splitCSV(s string) []string {
if strings.TrimSpace(s) == "" {
return nil
}
parts := strings.Split(s, ",")
for i := range parts {
parts[i] = strings.TrimSpace(parts[i])
}
// dedup
out := make([]string, 0, len(parts))
for _, p := range parts {
if p == "" || slices.Contains(out, p) {
continue
}
out = append(out, p)
}
return out
}
func getenvDefault(k, def string) string {
v := os.Getenv(k)
if v == "" {
return def
}
return v
}

316
internal/store/store.go
View File

@@ -1,316 +0,0 @@
package store
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"io"
"os"
"path/filepath"
"sort"
"strings"
"sync"
"time"
)
type Store struct {
blobDir string
metaDir string
tmpDir string
mu sync.RWMutex
}
type FileRecord struct {
ID string `json:"id"`
Name string `json:"name"`
Hash string `json:"hash"`
Size int64 `json:"size"`
Meta map[string]string `json:"meta,omitempty"`
CreatedAt time.Time `json:"createdAt"`
ContentType string `json:"contentType,omitempty"`
}
func (fr FileRecord) SafeName() string {
n := strings.TrimSpace(fr.Name)
if n == "" {
return fr.ID
}
return n
}
func Open(blobDir, metaDir, tmpDir string) (*Store, error) {
for _, p := range []string{blobDir, metaDir, tmpDir} {
if err := os.MkdirAll(p, 0o755); err != nil {
return nil, err
}
}
return &Store{blobDir: blobDir, metaDir: metaDir, tmpDir: tmpDir}, nil
}
func (s *Store) Put(ctx context.Context, r io.Reader, name, metaStr string) (*FileRecord, error) {
if name == "" {
name = "file"
}
tmp, err := os.CreateTemp(s.tmpDir, "upload-*")
if err != nil {
return nil, err
}
defer func() { tmp.Close(); os.Remove(tmp.Name()) }()
h := sha256.New()
n, err := io.Copy(io.MultiWriter(tmp, h), r)
if err != nil {
return nil, err
}
hash := hex.EncodeToString(h.Sum(nil))
blobPath := filepath.Join(s.blobDir, hash)
if _, err := os.Stat(blobPath); errors.Is(err, os.ErrNotExist) {
if err := os.Rename(tmp.Name(), blobPath); err != nil {
return nil, err
}
} else {
_ = os.Remove(tmp.Name())
}
rec := &FileRecord{
ID: newID(hash),
Name: name,
Hash: hash,
Size: n,
Meta: parseMeta(metaStr),
CreatedAt: time.Now().UTC(),
ContentType: "", // filled on GET via extension
}
if err := s.writeMeta(rec); err != nil {
return nil, err
}
return rec, nil
}
func (s *Store) Open(ctx context.Context, id string) (io.ReadSeekCloser, *FileRecord, error) {
rec, err := s.GetMeta(ctx, id)
if err != nil {
return nil, nil, err
}
f, err := os.Open(filepath.Join(s.blobDir, rec.Hash))
if err != nil {
return nil, nil, err
}
return f, rec, nil
}
func (s *Store) GetMeta(_ context.Context, id string) (*FileRecord, error) {
s.mu.RLock()
defer s.mu.RUnlock()
bb, err := os.ReadFile(filepath.Join(s.metaDir, id+".json"))
if err != nil {
return nil, err
}
var rec FileRecord
if err := json.Unmarshal(bb, &rec); err != nil {
return nil, err
}
return &rec, nil
}
func (s *Store) UpdateMeta(_ context.Context, id string, meta map[string]string) (*FileRecord, error) {
s.mu.Lock()
defer s.mu.Unlock()
path := filepath.Join(s.metaDir, id+".json")
bb, err := os.ReadFile(path)
if err != nil {
return nil, err
}
var rec FileRecord
if err := json.Unmarshal(bb, &rec); err != nil {
return nil, err
}
if rec.Meta == nil {
rec.Meta = map[string]string{}
}
for k, v := range meta {
rec.Meta[k] = v
}
nb, _ := json.Marshal(&rec)
if err := os.WriteFile(path, nb, 0o600); err != nil {
return nil, err
}
return &rec, nil
}
func (s *Store) Delete(_ context.Context, id string) error {
s.mu.Lock()
defer s.mu.Unlock()
// Only delete metadata; GC for unreferenced blobs is a separate task
return os.Remove(filepath.Join(s.metaDir, id+".json"))
}
func (s *Store) List(_ context.Context, q string, offset, limit int) ([]*FileRecord, int, error) {
if limit <= 0 || limit > 200 {
limit = 50
}
entries, err := os.ReadDir(s.metaDir)
if err != nil {
return nil, 0, err
}
var items []*FileRecord
for _, e := range entries {
if e.IsDir() || !strings.HasSuffix(e.Name(), ".json") {
continue
}
bb, err := os.ReadFile(filepath.Join(s.metaDir, e.Name()))
if err != nil {
continue
}
var rec FileRecord
if err := json.Unmarshal(bb, &rec); err != nil {
continue
}
if q == "" || strings.Contains(strings.ToLower(rec.Name), strings.ToLower(q)) {
items = append(items, &rec)
}
}
sort.Slice(items, func(i, j int) bool { return items[i].CreatedAt.After(items[j].CreatedAt) })
end := offset + limit
if offset > len(items) {
return []*FileRecord{}, 0, nil
}
if end > len(items) {
end = len(items)
}
next := 0
if end < len(items) {
next = end
}
return items[offset:end], next, nil
}
// --- Chunked uploads ---
type UploadSession struct {
ID string `json:"id"`
Name string `json:"name"`
Meta string `json:"meta"`
CreatedAt time.Time `json:"createdAt"`
}
func (s *Store) UploadInit(_ context.Context, name, meta string) (*UploadSession, error) {
id := newID(fmt.Sprintf("sess-%d", time.Now().UnixNano()))
us := &UploadSession{ID: id, Name: name, Meta: meta, CreatedAt: time.Now().UTC()}
// session file marker
if err := os.WriteFile(filepath.Join(s.tmpDir, id+".session"), []byte(name+""+meta), 0o600); err != nil {
return nil, err
}
return us, nil
}
func (s *Store) partPath(uid string, n int) string {
return filepath.Join(s.tmpDir, fmt.Sprintf("%s.part.%06d", uid, n))
}
func (s *Store) UploadPart(_ context.Context, uid string, n int, r io.Reader) error {
if _, err := os.Stat(filepath.Join(s.tmpDir, uid+".session")); err != nil {
return err
}
f, err := os.Create(s.partPath(uid, n))
if err != nil {
return err
}
defer f.Close()
_, err = io.Copy(f, r)
return err
}
func (s *Store) UploadComplete(ctx context.Context, uid string) (*FileRecord, error) {
if _, err := os.Stat(filepath.Join(s.tmpDir, uid+".session")); err != nil {
return nil, err
}
matches, _ := filepath.Glob(filepath.Join(s.tmpDir, uid+".part.*"))
if len(matches) == 0 {
return nil, errors.New("no parts uploaded")
}
sort.Strings(matches)
pr, pw := io.Pipe()
go func() {
for _, p := range matches {
f, err := os.Open(p)
if err != nil {
_ = pw.CloseWithError(err)
return
}
if _, err := io.Copy(pw, f); err != nil {
_ = pw.CloseWithError(err)
_ = f.Close()
return
}
_ = f.Close()
}
_ = pw.Close()
}()
// Read first line of session file for name/meta (simple format)
bb, _ := os.ReadFile(filepath.Join(s.tmpDir, uid+".session"))
lines := strings.SplitN(string(bb), "", 2)
name := "file"
meta := ""
if len(lines) >= 1 && strings.TrimSpace(lines[0]) != "" {
name = strings.TrimSpace(lines[0])
}
if len(lines) == 2 {
meta = strings.TrimSpace(lines[1])
}
rec, err := s.Put(ctx, pr, name, meta)
if err != nil {
return nil, err
}
for _, p := range matches {
_ = os.Remove(p)
}
_ = os.Remove(filepath.Join(s.tmpDir, uid+".session"))
return rec, nil
}
func (s *Store) UploadAbort(_ context.Context, uid string) error {
if _, err := os.Stat(filepath.Join(s.tmpDir, uid+".session")); err != nil {
return err
}
matches, _ := filepath.Glob(filepath.Join(s.tmpDir, uid+".part.*"))
for _, p := range matches {
_ = os.Remove(p)
}
return os.Remove(filepath.Join(s.tmpDir, uid+".session"))
}
// --- helpers ---
func (s *Store) writeMeta(rec *FileRecord) error {
s.mu.Lock()
defer s.mu.Unlock()
bb, _ := json.Marshal(rec)
return os.WriteFile(filepath.Join(s.metaDir, rec.ID+".json"), bb, 0o600)
}
func newID(seed string) string {
h := sha256.Sum256([]byte(fmt.Sprintf("%s|%d", seed, time.Now().UnixNano())))
return hex.EncodeToString(h[:16])
}
func parseMeta(s string) map[string]string {
if s == "" {
return nil
}
m := map[string]string{}
for _, kv := range strings.Split(s, ",") {
kvp := strings.SplitN(kv, "=", 2)
if len(kvp) == 2 {
m[strings.TrimSpace(kvp[0])] = strings.TrimSpace(kvp[1])
}
}
return m
}