feat(aether): initial commit – self-hosted Cloud, Phasen 1–7
Go-Backend (chi, pgx, golang-migrate) + React/TS-Frontend als PWA. Enthalten: Auth inkl. OIDC, Dateien mit Versionen und Papierkorb, Freigaben (Links + intern), WebDAV, Gruppenordner, Admin/Theming, Volltextsuche mit OCR, eigener PDF-Betrachter und WOPI-Office (Collabora/OnlyOffice). Nicht im Repo: deploy/.env (Sitzungsschlüssel, DB- und Admin-Passwort) sowie alle Daten- und Build-Verzeichnisse – siehe .gitignore. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
HoneyandClaude Opus 5 committed
commit
31a027c93a
225 files changed
+38042
No files matched your search
@@ -0,0 +1,315 @@
|
||||
package documents
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
// ExtractVersion ist die Fassung der Textaufbereitung.
|
||||
//
|
||||
// Sie steht an jedem Eintrag in file_text. Wird die Aufbereitung geändert, so
|
||||
// dass alte Einträge nicht mehr taugen, wird die Zahl erhöht – beim Start
|
||||
// werden sie dann erneut erkannt, statt still nach der alten Regel liegen zu
|
||||
// bleiben. Ohne so eine Marke fällt der Unterschied nie auf: Der Text ist ja
|
||||
// da, er kann nur weniger.
|
||||
//
|
||||
// 1 – Seitenumbruch (\f) bleibt im Text stehen (23.08.2026)
|
||||
const ExtractVersion = 1
|
||||
|
||||
// ErrNoJob heißt: gerade nichts zu tun. Kein Fehler, sondern der Normalfall.
|
||||
var ErrNoJob = errors.New("no pending job")
|
||||
|
||||
// maxAttempts begrenzt, wie oft ein Auftrag wiederholt wird.
|
||||
//
|
||||
// Zwei Wiederholungen fangen das ab, was vorübergehend war – ein Neustart
|
||||
// mitten im Lauf, kurzzeitig fehlender Speicherplatz. Was dreimal scheitert,
|
||||
// scheitert an der Datei selbst (kaputtes PDF, passwortgeschützt), und ein
|
||||
// vierter Versuch verbrennt nur Rechenzeit. Der Auftrag bleibt dann als
|
||||
// 'failed' mit Fehlertext stehen, statt still zu verschwinden.
|
||||
const maxAttempts = 3
|
||||
|
||||
type Store struct {
|
||||
db *pgxpool.Pool
|
||||
}
|
||||
|
||||
func NewStore(db *pgxpool.Pool) *Store {
|
||||
return &Store{db: db}
|
||||
}
|
||||
|
||||
// Job ist ein Auftrag samt allem, was der Arbeiter über die Datei braucht.
|
||||
type Job struct {
|
||||
ID uuid.UUID
|
||||
FileID uuid.UUID
|
||||
Attempts int
|
||||
Name string
|
||||
MimeType string
|
||||
StorageKey string
|
||||
Size int64
|
||||
}
|
||||
|
||||
// Enqueue trägt eine Datei zur Texterkennung ein.
|
||||
//
|
||||
// ON CONFLICT statt eines zweiten Eintrags: Wird dieselbe Datei erneut
|
||||
// hochgeladen, ist der alte Text hinfällig – der Auftrag wird auf 'pending'
|
||||
// zurückgesetzt, statt sich daneben zu stapeln. Der Zähler geht dabei auf 0,
|
||||
// weil eine neue Fassung eine neue Chance verdient: Vielleicht war es genau
|
||||
// der kaputte Scan, der ersetzt wurde.
|
||||
func (s *Store) Enqueue(ctx context.Context, fileID uuid.UUID) error {
|
||||
_, err := s.db.Exec(ctx, `
|
||||
INSERT INTO ocr_jobs (file_id, status)
|
||||
VALUES ($1, 'pending')
|
||||
ON CONFLICT (file_id) DO UPDATE
|
||||
SET status = 'pending', attempts = 0, last_error = '', updated_at = NOW()`,
|
||||
fileID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("enqueue ocr job: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Claim holt den ältesten offenen Auftrag und markiert ihn als laufend.
|
||||
//
|
||||
// FOR UPDATE SKIP LOCKED ist der Kern: Zwei Arbeiter greifen damit nie
|
||||
// denselben Auftrag – der zweite überspringt die gesperrte Zeile, statt darauf
|
||||
// zu warten. Heute läuft nur einer, aber die Sperre kostet nichts und erspart
|
||||
// die Suche nach doppelt erkannten Dateien, sobald es zwei werden.
|
||||
//
|
||||
// 'running' wird mit übernommen, nicht nur 'pending': Ein Auftrag, der beim
|
||||
// Neustart mittendrin war, steht sonst für immer auf 'running' und wird nie
|
||||
// wieder angefasst. Der Zähler sorgt dafür, dass daraus keine Endlosschleife
|
||||
// wird, falls die Datei den Server reproduzierbar umbringt.
|
||||
func (s *Store) Claim(ctx context.Context) (*Job, error) {
|
||||
var j Job
|
||||
err := s.db.QueryRow(ctx, `
|
||||
UPDATE ocr_jobs
|
||||
SET status = 'running', attempts = attempts + 1, updated_at = NOW()
|
||||
WHERE id = (
|
||||
SELECT j.id
|
||||
FROM ocr_jobs j
|
||||
JOIN files f ON f.id = j.file_id AND f.is_deleted = false
|
||||
WHERE j.status IN ('pending', 'running')
|
||||
AND j.attempts < $1
|
||||
ORDER BY j.created_at
|
||||
FOR UPDATE SKIP LOCKED
|
||||
LIMIT 1)
|
||||
RETURNING id, file_id, attempts,
|
||||
(SELECT name FROM files WHERE id = file_id),
|
||||
(SELECT mime_type FROM files WHERE id = file_id),
|
||||
(SELECT storage_key FROM files WHERE id = file_id),
|
||||
(SELECT size FROM files WHERE id = file_id)`,
|
||||
maxAttempts,
|
||||
).Scan(&j.ID, &j.FileID, &j.Attempts, &j.Name, &j.MimeType, &j.StorageKey, &j.Size)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return nil, ErrNoJob
|
||||
}
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("claim ocr job: %w", err)
|
||||
}
|
||||
return &j, nil
|
||||
}
|
||||
|
||||
// Complete speichert den Text und schließt den Auftrag ab. Beides in einer
|
||||
// Transaktion: Ein gespeicherter Text ohne erledigten Auftrag würde beim
|
||||
// nächsten Durchgang erneut erkannt, ein erledigter Auftrag ohne Text ließe die
|
||||
// Datei unauffindbar zurück.
|
||||
func (s *Store) Complete(ctx context.Context, job *Job, res Result) error {
|
||||
tx, err := s.db.Begin(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("begin: %w", err)
|
||||
}
|
||||
defer tx.Rollback(ctx)
|
||||
|
||||
if _, err := tx.Exec(ctx, `
|
||||
INSERT INTO file_text (file_id, content, source, lang, chars, pages, extract_version)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7)
|
||||
ON CONFLICT (file_id) DO UPDATE
|
||||
SET content = EXCLUDED.content, source = EXCLUDED.source,
|
||||
lang = EXCLUDED.lang, chars = EXCLUDED.chars,
|
||||
pages = EXCLUDED.pages, extract_version = EXCLUDED.extract_version,
|
||||
created_at = NOW()`,
|
||||
job.FileID, res.Text, res.Source, res.Lang, res.Chars, res.Pages, ExtractVersion,
|
||||
); err != nil {
|
||||
return fmt.Errorf("save text: %w", err)
|
||||
}
|
||||
|
||||
if _, err := tx.Exec(ctx,
|
||||
`UPDATE ocr_jobs SET status = 'done', last_error = '', updated_at = NOW() WHERE id = $1`,
|
||||
job.ID,
|
||||
); err != nil {
|
||||
return fmt.Errorf("finish job: %w", err)
|
||||
}
|
||||
return tx.Commit(ctx)
|
||||
}
|
||||
|
||||
// Fail vermerkt den Fehler. Ist der Zähler ausgeschöpft, bleibt der Auftrag als
|
||||
// 'failed' liegen; sonst geht er zurück in die Schlange.
|
||||
func (s *Store) Fail(ctx context.Context, job *Job, cause error) error {
|
||||
msg := cause.Error()
|
||||
if len(msg) > 500 {
|
||||
msg = msg[:500] + "…"
|
||||
}
|
||||
status := "pending"
|
||||
if job.Attempts >= maxAttempts {
|
||||
status = "failed"
|
||||
}
|
||||
_, err := s.db.Exec(ctx,
|
||||
`UPDATE ocr_jobs SET status = $2, last_error = $3, updated_at = NOW() WHERE id = $1`,
|
||||
job.ID, status, msg)
|
||||
return err
|
||||
}
|
||||
|
||||
// Skip schließt einen Auftrag ohne Text ab – etwa wenn die Datei inzwischen
|
||||
// gelöscht wurde. 'done' statt 'failed': Es ist nichts schiefgegangen.
|
||||
func (s *Store) Skip(ctx context.Context, jobID uuid.UUID, reason string) error {
|
||||
_, err := s.db.Exec(ctx,
|
||||
`UPDATE ocr_jobs SET status = 'done', last_error = $2, updated_at = NOW() WHERE id = $1`,
|
||||
jobID, reason)
|
||||
return err
|
||||
}
|
||||
|
||||
// Stats zählt die Aufträge je Zustand – für die Anzeige im Admin-Bereich und
|
||||
// für die Prüfung, ob der Arbeiter überhaupt vorankommt.
|
||||
type Stats struct {
|
||||
Pending int `json:"pending"`
|
||||
Running int `json:"running"`
|
||||
Done int `json:"done"`
|
||||
Failed int `json:"failed"`
|
||||
// Indexed ist die Zahl der Dateien mit gespeichertem Text.
|
||||
Indexed int `json:"indexed"`
|
||||
}
|
||||
|
||||
func (s *Store) Stats(ctx context.Context) (Stats, error) {
|
||||
var st Stats
|
||||
err := s.db.QueryRow(ctx, `
|
||||
SELECT COUNT(*) FILTER (WHERE status = 'pending'),
|
||||
COUNT(*) FILTER (WHERE status = 'running'),
|
||||
COUNT(*) FILTER (WHERE status = 'done'),
|
||||
COUNT(*) FILTER (WHERE status = 'failed'),
|
||||
(SELECT COUNT(*) FROM file_text)
|
||||
FROM ocr_jobs`,
|
||||
).Scan(&st.Pending, &st.Running, &st.Done, &st.Failed, &st.Indexed)
|
||||
if err != nil {
|
||||
return st, fmt.Errorf("ocr stats: %w", err)
|
||||
}
|
||||
return st, nil
|
||||
}
|
||||
|
||||
// StatsForUser zählt nur, was diesem Nutzer gehört.
|
||||
//
|
||||
// Warum es das getrennt von Stats gibt: Den Gesamtstand darf nur der
|
||||
// Administrator sehen – die Zahl aller Dokumente auf dem Server verrät etwas
|
||||
// über fremde Bestände. Wissen muss es aber genau der, der die Rechnung
|
||||
// hochgeladen hat: Findet seine Suche sie nicht, ist die Frage „ist sie schon
|
||||
// gelesen?" die erste, die er stellt.
|
||||
//
|
||||
// Gezählt wird über owner_id, also die eigenen Dateien. Dateien in
|
||||
// Gruppenordnern bleiben außen vor – sie gehören der Gruppe, und ihr Stand
|
||||
// wäre für den Einzelnen keine Auskunft über den eigenen Bestand.
|
||||
func (s *Store) StatsForUser(ctx context.Context, userID uuid.UUID) (Stats, error) {
|
||||
var st Stats
|
||||
err := s.db.QueryRow(ctx, `
|
||||
SELECT COUNT(*) FILTER (WHERE j.status = 'pending'),
|
||||
COUNT(*) FILTER (WHERE j.status = 'running'),
|
||||
COUNT(*) FILTER (WHERE j.status = 'done'),
|
||||
COUNT(*) FILTER (WHERE j.status = 'failed'),
|
||||
COUNT(*) FILTER (WHERE t.file_id IS NOT NULL)
|
||||
FROM files f
|
||||
JOIN ocr_jobs j ON j.file_id = f.id
|
||||
LEFT JOIN file_text t ON t.file_id = f.id
|
||||
WHERE f.owner_id = $1 AND f.is_deleted = false`,
|
||||
userID,
|
||||
).Scan(&st.Pending, &st.Running, &st.Done, &st.Failed, &st.Indexed)
|
||||
if err != nil {
|
||||
return st, fmt.Errorf("ocr stats for user: %w", err)
|
||||
}
|
||||
return st, nil
|
||||
}
|
||||
|
||||
// EnqueueMissing trägt alle Bestandsdateien nach, die noch keinen Auftrag
|
||||
// haben. Ohne das bliebe alles unauffindbar, was vor dem Einbau hochgeladen
|
||||
// wurde – und das ist beim ersten Start genau der gesamte Bestand.
|
||||
//
|
||||
// Die Auswahl der Typen passiert hier in SQL statt über Supported(): Bei
|
||||
// zehntausend Dateien wäre es Unfug, jede einzelne nach Go zu holen, nur um
|
||||
// ihren MIME-Typ anzusehen.
|
||||
func (s *Store) EnqueueMissing(ctx context.Context, limit int) (int, error) {
|
||||
tag, err := s.db.Exec(ctx, `
|
||||
INSERT INTO ocr_jobs (file_id, status)
|
||||
SELECT f.id, 'pending'
|
||||
FROM files f
|
||||
WHERE f.is_deleted = false
|
||||
AND (f.mime_type ILIKE '%pdf%'
|
||||
OR (f.mime_type ILIKE 'image/%' AND f.mime_type NOT ILIKE '%svg%')
|
||||
OR f.name ~* '\.(pdf|png|jpe?g|tiff?|bmp|webp)$')
|
||||
AND NOT EXISTS (SELECT 1 FROM ocr_jobs j WHERE j.file_id = f.id)
|
||||
ORDER BY f.created_at DESC
|
||||
LIMIT $1
|
||||
ON CONFLICT (file_id) DO NOTHING`, limit)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("enqueue missing: %w", err)
|
||||
}
|
||||
return int(tag.RowsAffected()), nil
|
||||
}
|
||||
|
||||
// EnqueueOutdated legt Einträge erneut in die Warteschlange, die nach einer
|
||||
// älteren Regel entstanden sind.
|
||||
//
|
||||
// Anlass war die Seitenmarke: Bis Fassung 1 wurde der Seitenumbruch aus dem
|
||||
// Text entfernt. Wer damit indiziert wurde, ist zwar auffindbar – aber der
|
||||
// Betrachter kann bei einem Scan nicht sagen, auf welcher Seite das Wort
|
||||
// steht. Ein stiller Rest, der sich nur durch erneutes Erkennen auflöst.
|
||||
//
|
||||
// Bewusst mit derselben Obergrenze wie EnqueueMissing: Bei einem großen
|
||||
// Bestand soll ein Neustart nicht die ganze Platte an OCR verheizen, sondern
|
||||
// den Rückstand über mehrere Starte abtragen.
|
||||
func (s *Store) EnqueueOutdated(ctx context.Context, limit int) (int, error) {
|
||||
tag, err := s.db.Exec(ctx, `
|
||||
INSERT INTO ocr_jobs (file_id, status)
|
||||
SELECT t.file_id, 'pending'
|
||||
FROM file_text t
|
||||
JOIN files f ON f.id = t.file_id AND f.is_deleted = false
|
||||
WHERE t.extract_version < $2
|
||||
LIMIT $1
|
||||
ON CONFLICT (file_id) DO UPDATE
|
||||
SET status = 'pending', attempts = 0, last_error = '', updated_at = NOW()`,
|
||||
limit, ExtractVersion)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("enqueue outdated: %w", err)
|
||||
}
|
||||
return int(tag.RowsAffected()), nil
|
||||
}
|
||||
|
||||
// TextPages liefert den erkannten Text einer Datei, in Seiten zerlegt.
|
||||
//
|
||||
// Für den PDF-Betrachter: Bei einem Scan steht im PDF selbst kein Wort, nur
|
||||
// ein Bild. Ohne diesen Weg fände die Suche im Dokument nichts – ausgerechnet
|
||||
// bei dem Dateityp, für den der erkannte Text überhaupt gebaut wurde.
|
||||
//
|
||||
// Die Zugriffsprüfung passiert im Aufrufer; hier liegt nur die Abfrage.
|
||||
func (s *Store) TextPages(ctx context.Context, fileID uuid.UUID) ([]string, string, error) {
|
||||
var content, source string
|
||||
err := s.db.QueryRow(ctx,
|
||||
`SELECT content, source FROM file_text WHERE file_id = $1`, fileID,
|
||||
).Scan(&content, &source)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return nil, "", nil
|
||||
}
|
||||
if err != nil {
|
||||
return nil, "", fmt.Errorf("text pages: %w", err)
|
||||
}
|
||||
return strings.Split(content, "\f"), source, nil
|
||||
}
|
||||
|
||||
// DeleteText entfernt den Text einer Datei. Wird beim endgültigen Löschen
|
||||
// gebraucht; solange die Datei nur im Papierkorb liegt, bleibt er stehen –
|
||||
// beim Wiederherstellen wäre er sonst weg.
|
||||
func (s *Store) DeleteText(ctx context.Context, fileID uuid.UUID) error {
|
||||
_, err := s.db.Exec(ctx, `DELETE FROM file_text WHERE file_id = $1`, fileID)
|
||||
return err
|
||||
}
|
||||
Reference in new issue
Block a user