kjv_api/internal/kjv/logdb.go

package kjv

import (
	"database/sql"
	"fmt"
	"os"
	"strings"
	"sync"
	"syscall"
	"time"
	"unicode/utf8"
)

// DBLog writes log lines into the logs table of the companion database. Lines
// are queued and inserted in batches so requests never wait on a write. Until
// a database is attached (and if it turns out to be read-only) lines go to stderr.
type DBLog struct {
	mu      sync.Mutex
	db      *sql.DB
	ch      chan logLine
	done    chan struct{}
	started bool
	retain  int
}

type logLine struct {
	ts, level, msg string
}

var lg = &DBLog{ch: make(chan logLine, 4096), done: make(chan struct{})}

const (
	logTimeFmt = "2006-01-02 15:04:05.000"
	maxLogLine = 400 // request URLs and queries come from strangers
)

func truncate(s string, n int) string {
	if len(s) <= n {
		return s
	}
	for n > 0 && !utf8.RuneStart(s[n]) { // don't cut a character in half
		n--
	}
	return s[:n] + "..."
}

func (l *DBLog) Attach(db *sql.DB, retainDays int) {
	if _, err := db.Exec(`CREATE TABLE IF NOT EXISTS logs (id INTEGER PRIMARY KEY AUTOINCREMENT, ts TEXT NOT NULL, level TEXT NOT NULL, msg TEXT NOT NULL)`); err != nil {
		fmt.Fprintln(os.Stderr, "WARN: cannot write logs to the database, logging to stderr:", err)
		return
	}
	l.retain = retainDays
	l.prune(db)
	l.mu.Lock()
	l.db, l.started = db, true
	l.mu.Unlock()
	go l.run()
}

func (l *DBLog) prune(db *sql.DB) {
	db.Exec("DELETE FROM logs WHERE ts < ?", time.Now().AddDate(0, 0, -l.retain).Format(logTimeFmt))
}

func (l *DBLog) run() {
	tick := time.NewTicker(time.Second)
	defer tick.Stop()
	daily := time.NewTicker(6 * time.Hour)
	defer daily.Stop()
	var batch []logLine
	flush := func() {
		if len(batch) == 0 {
			return
		}
		tx, err := l.db.Begin()
		if err == nil {
			for _, ln := range batch {
				if _, err = tx.Exec("INSERT INTO logs (ts, level, msg) VALUES (?, ?, ?)", ln.ts, ln.level, ln.msg); err != nil {
					break
				}
			}
			if err == nil {
				err = tx.Commit()
			} else {
				tx.Rollback()
			}
		}
		if err != nil {
			for _, ln := range batch {
				fmt.Fprintf(os.Stderr, "%s %s %s\n", ln.ts, ln.level, ln.msg)
			}
		}
		batch = batch[:0]
	}
	for {
		select {
		case ln := <-l.ch:
			batch = append(batch, ln)
			if len(batch) >= 200 {
				flush()
			}
		case <-tick.C:
			flush()
		case <-daily.C:
			l.prune(l.db)
		case <-l.done:
			for {
				select {
				case ln := <-l.ch:
					batch = append(batch, ln)
				default:
					flush()
					l.done <- struct{}{}
					return
				}
			}
		}
	}
}

// Close flushes queued lines.
func (l *DBLog) Close() {
	l.mu.Lock()
	started := l.started
	l.mu.Unlock()
	if !started {
		return
	}
	l.done <- struct{}{}
	<-l.done
}

// Under systemd (stderr is the journal) or with KJV_LOG_STDERR=1 every line is
// also written to stderr, where the journal collects it. The "<n>" prefix is
// the syslog priority journald understands; it adds its own timestamps.
var teeStderr = stderrIsJournal() || os.Getenv("KJV_LOG_STDERR") == "1"
var journalPriority = map[string]int{"ERROR": 3, "WARN": 4, "INFO": 6}

// stderrIsJournal: JOURNAL_STREAM is inherited by everything a systemd
// started desktop session runs, terminals included, so it only counts when it
// names the device:inode stderr really is (as systemd.exec(5) advises).
func stderrIsJournal() bool {
	want := os.Getenv("JOURNAL_STREAM")
	if want == "" {
		return false
	}
	fi, err := os.Stderr.Stat()
	if err != nil {
		return false
	}
	st, ok := fi.Sys().(*syscall.Stat_t)
	return ok && fmt.Sprintf("%d:%d", st.Dev, st.Ino) == want
}

func (l *DBLog) log(level, format string, args ...any) {
	ln := logLine{time.Now().Format(logTimeFmt), level, truncate(fmt.Sprintf(format, args...), maxLogLine)}
	if teeStderr {
		fmt.Fprintf(os.Stderr, "<%d>%s\n", journalPriority[level], strings.ReplaceAll(ln.msg, "\n", " "))
	}
	l.mu.Lock()
	started := l.started
	l.mu.Unlock()
	if started {
		select {
		case l.ch <- ln:
			return
		default: // queue full: don't block a request
		}
	}
	if !teeStderr { // not in the database and not already on stderr
		fmt.Fprintf(os.Stderr, "%s %s %s\n", ln.ts, ln.level, ln.msg)
	}
}

func (l *DBLog) Info(format string, args ...any)  { l.log("INFO", format, args...) }
func (l *DBLog) Warn(format string, args ...any)  { l.log("WARN", format, args...) }
func (l *DBLog) Error(format string, args ...any) { l.log("ERROR", format, args...) }

// ShowLogs prints the newest n log lines (oldest first).
func ShowLogs(dbPath string, n int) error {
	db, err := sql.Open("sqlite", "file:"+dbPath+"?mode=ro&_pragma=busy_timeout(5000)")
	if err != nil {
		return err
	}
	defer db.Close()
	rows, err := db.Query("SELECT ts, level, msg FROM (SELECT * FROM logs ORDER BY id DESC LIMIT ?) ORDER BY id", n)
	if err != nil {
		return err
	}
	defer rows.Close()
	for rows.Next() {
		var ts, level, msg string
		if err := rows.Scan(&ts, &level, &msg); err != nil {
			return err
		}
		fmt.Printf("%s %-5s %s\n", ts, level, msg)
	}
	return rows.Err()
}