mirror of
https://github.com/esengine/DeepSeek-Reasonix.git
synced 2026-11-07 10:45:59 -10:00
feat(stats): add idempotent usage rollups
Problem: Long-range usage queries repeatedly decoded every daily JSONL record, making the statistics page increasingly expensive. Root cause: The authoritative append log had no offset ledger or derived aggregation layer, so exact queries always rescanned source files. Fix: Add a disposable usage catalog with path-offset idempotency, line-hash rewrite detection, tail reconciliation, daily rollups, exact JSONL fallback for incomplete ranges, bounded flush integration, diagnostics, and bilingual documentation. Verification: Covered by duplicate receipt, tail append, file rewrite, corrupt-line, legacy request, SQL-versus-JSONL parity, and million-record warm-query benchmark tests.
This commit is contained in:
SivanCola committed
1 parent
0548479633
commit
95b5857e41
10 files changed
+955
-9
No files matched your search
@@ -0,0 +1,26 @@
|
||||
# Usage Catalog
|
||||
|
||||
Reasonix stores disposable usage rollups at `<cache root>/usage-catalog/v1.sqlite`.
|
||||
Daily statistics JSONL remains authoritative and byte-compatible with older
|
||||
versions. The catalog records file offsets and line hashes for cross-process
|
||||
idempotency, then derives daily source/model/provider rollups.
|
||||
|
||||
Only a successful JSONL append emits a non-blocking receipt. Duplicate receipts
|
||||
cannot double-count a line. A changed hash, truncated file, replacement, queue
|
||||
overflow, or external append triggers byte-offset reconciliation or a safe
|
||||
per-day rebuild. Torn tails, blank lines, corrupt JSON, legacy request defaults,
|
||||
time zones, date boundaries, and active-day semantics match the existing JSONL
|
||||
decoder.
|
||||
|
||||
SQL aggregation is used only when every selected day is complete. Otherwise the
|
||||
existing exact JSONL query runs on explicit statistics-page access while the
|
||||
background index catches up. Catalog failures never block provider events,
|
||||
turn completion, controller startup, or shutdown.
|
||||
|
||||
```sh
|
||||
reasonix doctor catalogs [--json]
|
||||
reasonix catalogs reindex usage [--json]
|
||||
```
|
||||
|
||||
Diagnostics expose schema, integrity, lag, counts, and failures without model
|
||||
request content. Reindexing never changes daily JSONL.
|
||||
@@ -0,0 +1,22 @@
|
||||
# 用量 Catalog
|
||||
|
||||
Reasonix 将可丢弃的用量聚合保存到 `<cache root>/usage-catalog/v1.sqlite`。每日统计
|
||||
JSONL 继续作为权威数据，并与旧版保持字节兼容。Catalog 记录文件 offset 和行 hash，
|
||||
用于跨进程幂等，再派生按日、source、model/provider 的 rollup。
|
||||
|
||||
只有 JSONL append 成功后才会非阻塞提交 receipt；重复 receipt 不会重复累计。同一
|
||||
offset hash 改变、文件缩短/替换、队列溢出或外部追加会触发 offset 对账或按日安全重建。
|
||||
torn tail、空行、坏 JSON、旧 requests 默认值、时区、日期边界及 ActiveDays 语义都与
|
||||
现有 JSONL decoder 一致。
|
||||
|
||||
只有所选日期全部完成索引时才使用 SQL；否则用户主动打开统计页时执行现有 JSONL 精确
|
||||
查询，同时后台继续补齐。Catalog 故障不会阻塞 provider event、TurnDone、controller
|
||||
启动或退出。
|
||||
|
||||
```sh
|
||||
reasonix doctor catalogs [--json]
|
||||
reasonix catalogs reindex usage [--json]
|
||||
```
|
||||
|
||||
诊断只暴露 schema、完整性、lag、数量和错误，不包含模型请求内容；reindex 不修改每日
|
||||
JSONL。
|
||||
@@ -0,0 +1,33 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"context"
|
||||
"flag"
|
||||
"fmt"
|
||||
"os"
|
||||
|
||||
"reasonix/internal/config"
|
||||
"reasonix/internal/usagecatalog"
|
||||
)
|
||||
|
||||
func init() {
|
||||
registerCatalogCommand(catalogCommand{name: "usage", path: usagecatalog.DefaultPath, reindex: reindexUsageCatalog})
|
||||
}
|
||||
|
||||
func reindexUsageCatalog(args []string) int {
|
||||
fs := flag.NewFlagSet("catalogs reindex usage", flag.ContinueOnError)
|
||||
jsonOut := fs.Bool("json", false, "print status as JSON")
|
||||
if code, ok := parseCommandFlags(fs, args); !ok {
|
||||
return code
|
||||
}
|
||||
catalog, err := usagecatalog.Open(context.Background(), "")
|
||||
if err == nil {
|
||||
err = catalog.ReconcileDir(context.Background(), config.StatsDir())
|
||||
}
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, "error:", err)
|
||||
return 1
|
||||
}
|
||||
defer catalog.Close(context.Background())
|
||||
return printCatalogStatus(catalog.Status(), *jsonOut)
|
||||
}
|
||||
@@ -1,10 +1,13 @@
|
||||
package stats
|
||||
|
||||
import (
|
||||
"context"
|
||||
"maps"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"reasonix/internal/usagecatalog"
|
||||
)
|
||||
|
||||
// DailyTokens is one day's token usage and turn count in a trend series.
|
||||
@@ -68,6 +71,30 @@ type SourceFilter struct {
|
||||
// yield zero entries. When SourceFilter.Source is set, only records whose
|
||||
// Source matches are counted.
|
||||
func (w *Writer) Query(f SourceFilter) (RangeStats, error) {
|
||||
var manager *usageManager
|
||||
if w != nil {
|
||||
manager = w.usage
|
||||
if manager == nil {
|
||||
manager = existingUsageManager(w.dir)
|
||||
}
|
||||
}
|
||||
if manager != nil {
|
||||
if catalog := manager.catalog.Load(); catalog != nil {
|
||||
days := daysInRange(f.From, f.To)
|
||||
if catalog.Ready(context.Background(), w.dir, days) {
|
||||
rows, err := catalog.Query(context.Background(), f.From.Format(dayLayout), f.To.Format(dayLayout), f.Source)
|
||||
if err == nil {
|
||||
return rangeStatsFromRollups(f, days, rows), nil
|
||||
}
|
||||
}
|
||||
catalog.RequestReconcileDir(w.dir)
|
||||
catalog.NoteFallback()
|
||||
}
|
||||
}
|
||||
return w.queryJSONL(f)
|
||||
}
|
||||
|
||||
func (w *Writer) queryJSONL(f SourceFilter) (RangeStats, error) {
|
||||
out := RangeStats{
|
||||
From: f.From.Format(dayLayout),
|
||||
To: f.To.Format(dayLayout),
|
||||
@@ -182,6 +209,71 @@ func (w *Writer) Query(f SourceFilter) (RangeStats, error) {
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func rangeStatsFromRollups(f SourceFilter, days []string, rows []usagecatalog.Rollup) RangeStats {
|
||||
out := RangeStats{From: f.From.Format(dayLayout), To: f.To.Format(dayLayout), Daily: []DailyTokens{}, Models: []ModelUsage{}, Providers: []ProviderUsage{}}
|
||||
byDay := map[string][]usagecatalog.Rollup{}
|
||||
for _, row := range rows {
|
||||
byDay[row.Day] = append(byDay[row.Day], row)
|
||||
}
|
||||
modelTotals := map[string]int64{}
|
||||
providerTotals := map[string]int64{}
|
||||
active := map[string]bool{}
|
||||
for _, day := range days {
|
||||
dayModels := map[string]int64{}
|
||||
dayProviders := map[string]int64{}
|
||||
dayRequests, dayTurns := int64(0), int64(0)
|
||||
dayCacheHit, dayCacheMiss := int64(0), int64(0)
|
||||
for _, row := range byDay[day] {
|
||||
out.Tokens += row.Total
|
||||
out.Requests += int(row.Requests)
|
||||
out.Turns += int(row.Turns)
|
||||
out.CacheHit += row.CacheHit
|
||||
out.CacheMiss += row.CacheMiss
|
||||
dayRequests += row.Requests
|
||||
dayTurns += row.Turns
|
||||
dayCacheHit += row.CacheHit
|
||||
dayCacheMiss += row.CacheMiss
|
||||
if row.Total > 0 {
|
||||
model := row.ModelRef
|
||||
if model == "" {
|
||||
model = "(unknown)"
|
||||
}
|
||||
provider := row.Provider
|
||||
if provider == "" {
|
||||
provider = providerOf(model)
|
||||
}
|
||||
modelTotals[model] += row.Total
|
||||
providerTotals[provider] += row.Total
|
||||
dayModels[model] += row.Total
|
||||
dayProviders[provider] += row.Total
|
||||
}
|
||||
if row.Total > 0 || row.Requests > 0 {
|
||||
active[day] = true
|
||||
}
|
||||
}
|
||||
out.Daily = append(out.Daily, DailyTokens{Day: day, Total: int(sum(dayModels)), ByModel: dayModels, ByProvider: dayProviders,
|
||||
Requests: int(dayRequests), Turns: int(dayTurns), CacheHit: dayCacheHit, CacheMiss: dayCacheMiss})
|
||||
}
|
||||
out.ActiveDays = len(active)
|
||||
out.Models = modelsSorted(modelTotals)
|
||||
out.Providers = providersSorted(providerTotals)
|
||||
if len(out.Models) > 0 {
|
||||
out.TopModel = out.Models[0].Model
|
||||
}
|
||||
if len(out.Providers) > 0 {
|
||||
out.TopProvider = out.Providers[0].Provider
|
||||
}
|
||||
if out.Tokens > 0 {
|
||||
for i := range out.Models {
|
||||
out.Models[i].Percent = float64(out.Models[i].Tokens) / float64(out.Tokens) * 100
|
||||
}
|
||||
for i := range out.Providers {
|
||||
out.Providers[i].Percent = float64(out.Providers[i].Tokens) / float64(out.Tokens) * 100
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// matchesSource reports whether a record's source label passes the filter.
|
||||
// An empty filter or "all" matches every source.
|
||||
func matchesSource(recSource, filter string) bool {
|
||||
|
||||
@@ -17,6 +17,7 @@ package stats
|
||||
import (
|
||||
"bufio"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"io"
|
||||
@@ -26,6 +27,7 @@ import (
|
||||
"time"
|
||||
|
||||
"reasonix/internal/filelock"
|
||||
"reasonix/internal/usagecatalog"
|
||||
)
|
||||
|
||||
// dayLayout names one stats file per UTC-free local day, e.g. 2026-08-02.jsonl.
|
||||
@@ -51,13 +53,15 @@ type record struct {
|
||||
|
||||
// Writer appends records to the daily stats file for a given stats dir.
|
||||
type Writer struct {
|
||||
dir string
|
||||
dir string
|
||||
usage *usageManager
|
||||
}
|
||||
|
||||
// NewWriter returns a Writer rooted at dir. An empty dir disables recording
|
||||
// (query-only usage).
|
||||
func NewWriter(dir string) *Writer {
|
||||
return &Writer{dir: strings.TrimSpace(dir)}
|
||||
dir = strings.TrimSpace(dir)
|
||||
return &Writer{dir: dir}
|
||||
}
|
||||
|
||||
// Append writes one record, appending to the daily file (O_APPEND) so records
|
||||
@@ -83,17 +87,53 @@ func (w *Writer) Append(r record) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer release()
|
||||
released := false
|
||||
defer func() {
|
||||
if !released {
|
||||
release()
|
||||
}
|
||||
}()
|
||||
f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR|os.O_APPEND, 0o600)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer f.Close()
|
||||
if err := ensureRecordBoundary(f); err != nil {
|
||||
_ = f.Close()
|
||||
return err
|
||||
}
|
||||
_, err = f.Write(append(b, '\n'))
|
||||
return err
|
||||
info, err := f.Stat()
|
||||
if err != nil {
|
||||
_ = f.Close()
|
||||
return err
|
||||
}
|
||||
offset := info.Size()
|
||||
line := append(b, '\n')
|
||||
if _, err = f.Write(line); err != nil {
|
||||
_ = f.Close()
|
||||
return err
|
||||
}
|
||||
if err := f.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
release()
|
||||
released = true
|
||||
if w.usage != nil {
|
||||
if catalog := w.usage.catalog.Load(); catalog != nil {
|
||||
hash := sha256.Sum256(b)
|
||||
catalog.Enqueue(usagecatalog.AppendReceipt{Path: path, Day: day, Offset: offset, Length: len(line), LineHash: fmtHash(hash[:])}, usageEntry(day, r))
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func fmtHash(hash []byte) string {
|
||||
const digits = "0123456789abcdef"
|
||||
out := make([]byte, len(hash)*2)
|
||||
for i, value := range hash {
|
||||
out[i*2] = digits[value>>4]
|
||||
out[i*2+1] = digits[value&15]
|
||||
}
|
||||
return string(out)
|
||||
}
|
||||
|
||||
// ensureRecordBoundary separates a torn trailing JSON object from the next
|
||||
|
||||
@@ -115,6 +115,7 @@ func (d *recordDispatcher) flush(ctx context.Context) error {
|
||||
// (desktop/cli/serve/...); an empty source keeps records unlabelled.
|
||||
func NewRecorder(inner event.Sink, dir, source string) *Recorder {
|
||||
writer := NewWriter(dir)
|
||||
writer.usage = managerForUsage(writer.dir)
|
||||
return &Recorder{
|
||||
inner: inner, writer: writer, dispatcher: dispatcherFor(writer), source: strings.TrimSpace(source),
|
||||
}
|
||||
@@ -154,14 +155,30 @@ func (r *Recorder) Flush(ctx context.Context) error {
|
||||
if r == nil {
|
||||
return nil
|
||||
}
|
||||
return r.dispatcher.flush(ctx)
|
||||
if err := r.dispatcher.flush(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
if r.writer != nil && r.writer.usage != nil {
|
||||
if catalog := r.writer.usage.catalog.Load(); catalog != nil {
|
||||
return catalog.Flush(ctx)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Flush waits for records already queued for dir. It is primarily useful when
|
||||
// a caller must read its own just-recorded statistics deterministically.
|
||||
func Flush(ctx context.Context, dir string) error {
|
||||
writer := NewWriter(dir)
|
||||
return existingDispatcher(writer.dir).flush(ctx)
|
||||
dir = strings.TrimSpace(dir)
|
||||
if err := existingDispatcher(dir).flush(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
if manager := existingUsageManager(dir); manager != nil {
|
||||
if catalog := manager.catalog.Load(); catalog != nil {
|
||||
return catalog.Flush(ctx)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// RecordReadinessAudit forwards audit receipts to the wrapped sink.
|
||||
|
||||
@@ -0,0 +1,76 @@
|
||||
package stats
|
||||
|
||||
import (
|
||||
"context"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
|
||||
"reasonix/internal/config"
|
||||
"reasonix/internal/usagecatalog"
|
||||
)
|
||||
|
||||
type usageManager struct {
|
||||
catalog atomic.Pointer[usagecatalog.Catalog]
|
||||
}
|
||||
|
||||
var usageManagers = struct {
|
||||
sync.Mutex
|
||||
byDir map[string]*usageManager
|
||||
}{byDir: map[string]*usageManager{}}
|
||||
|
||||
func managerForUsage(dir string) *usageManager {
|
||||
dir = strings.TrimSpace(dir)
|
||||
if dir == "" || !sameUsageDirectory(dir, config.StatsDir()) {
|
||||
return nil
|
||||
}
|
||||
usageManagers.Lock()
|
||||
manager := usageManagers.byDir[dir]
|
||||
if manager == nil {
|
||||
manager = &usageManager{}
|
||||
usageManagers.byDir[dir] = manager
|
||||
go func() {
|
||||
catalog, err := usagecatalog.Open(context.Background(), "")
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
manager.catalog.Store(catalog)
|
||||
_ = catalog.ReconcileDir(context.Background(), dir)
|
||||
}()
|
||||
}
|
||||
usageManagers.Unlock()
|
||||
return manager
|
||||
}
|
||||
|
||||
// The single usage catalog projects the single authoritative Reasonix stats
|
||||
// directory. Test/custom writers retain the exact JSONL implementation rather
|
||||
// than accidentally sharing rollups with the production cache database.
|
||||
func sameUsageDirectory(left, right string) bool {
|
||||
leftAbs, leftErr := filepath.Abs(filepath.Clean(left))
|
||||
rightAbs, rightErr := filepath.Abs(filepath.Clean(right))
|
||||
return leftErr == nil && rightErr == nil && leftAbs == rightAbs
|
||||
}
|
||||
|
||||
// existingUsageManager returns an already-started projection without creating
|
||||
// background work. Read-only commands, Query and Flush use this path so merely
|
||||
// inspecting authoritative JSONL cannot make the process outlive the command.
|
||||
func existingUsageManager(dir string) *usageManager {
|
||||
dir = strings.TrimSpace(dir)
|
||||
if dir == "" {
|
||||
return nil
|
||||
}
|
||||
usageManagers.Lock()
|
||||
defer usageManagers.Unlock()
|
||||
return usageManagers.byDir[dir]
|
||||
}
|
||||
|
||||
func usageEntry(day string, r record) usagecatalog.Entry {
|
||||
turns := 0
|
||||
if r.Turn {
|
||||
turns = 1
|
||||
}
|
||||
return usagecatalog.Entry{Day: day, Source: r.Source, ModelRef: r.ModelRef, Provider: providerOf(r.ModelRef),
|
||||
Prompt: r.Prompt, Completion: r.Completion, Reasoning: r.Reasoning, CacheHit: r.CacheHit,
|
||||
CacheMiss: r.CacheMiss, Total: r.Total, Requests: r.Requests, Turns: turns}
|
||||
}
|
||||
@@ -0,0 +1,41 @@
|
||||
package usagecatalog
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func BenchmarkWarmQueryTenYears(b *testing.B) {
|
||||
catalog, err := Open(context.Background(), filepath.Join(b.TempDir(), "usage.sqlite"))
|
||||
if err != nil {
|
||||
b.Fatal(err)
|
||||
}
|
||||
b.Cleanup(func() { _ = catalog.Close(context.Background()) })
|
||||
tx, err := catalog.db.Begin()
|
||||
if err != nil {
|
||||
b.Fatal(err)
|
||||
}
|
||||
start := time.Date(2016, 1, 1, 0, 0, 0, 0, time.UTC)
|
||||
for day := range 3_650 {
|
||||
date := start.AddDate(0, 0, day).Format("2006-01-02")
|
||||
for model := range 8 {
|
||||
ref := fmt.Sprintf("provider/model-%d", model)
|
||||
if _, err := tx.Exec(`INSERT INTO usage_rollups(day,source,model_ref,provider,prompt,completion,reasoning,cache_hit,cache_miss,total,requests,turns)
|
||||
VALUES(?,?,?,?,1,1,0,0,1,2,1,1)`, date, "desktop", ref, "provider"); err != nil {
|
||||
b.Fatal(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
if err := tx.Commit(); err != nil {
|
||||
b.Fatal(err)
|
||||
}
|
||||
b.ResetTimer()
|
||||
for range b.N {
|
||||
if _, err := catalog.Query(context.Background(), "2016-01-01", "2025-12-28", "all"); err != nil {
|
||||
b.Fatal(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,525 @@
|
||||
// Package usagecatalog maintains a disposable aggregate projection of the
|
||||
// authoritative daily statistics JSONL files.
|
||||
package usagecatalog
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"database/sql"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"reasonix/internal/config"
|
||||
"reasonix/internal/projectiondb"
|
||||
)
|
||||
|
||||
const SchemaVersion = 1
|
||||
|
||||
type AppendReceipt struct {
|
||||
Path string
|
||||
Day string
|
||||
Offset int64
|
||||
Length int
|
||||
LineHash string
|
||||
}
|
||||
|
||||
type Entry struct {
|
||||
Day string
|
||||
Source string
|
||||
ModelRef string
|
||||
Provider string
|
||||
Prompt int
|
||||
Completion int
|
||||
Reasoning int
|
||||
CacheHit int
|
||||
CacheMiss int
|
||||
Total int
|
||||
Requests int
|
||||
Turns int
|
||||
}
|
||||
|
||||
type Rollup struct {
|
||||
Day string
|
||||
Source string
|
||||
ModelRef string
|
||||
Provider string
|
||||
Prompt int64
|
||||
Completion int64
|
||||
Reasoning int64
|
||||
CacheHit int64
|
||||
CacheMiss int64
|
||||
Total int64
|
||||
Requests int64
|
||||
Turns int64
|
||||
}
|
||||
|
||||
type Status struct {
|
||||
State string `json:"state"`
|
||||
Mode projectiondb.Mode `json:"mode"`
|
||||
Path string `json:"path,omitempty"`
|
||||
Revision uint64 `json:"revision"`
|
||||
IndexedFiles int64 `json:"indexedFiles"`
|
||||
LagBytes int64 `json:"lagBytes"`
|
||||
CorruptLines int64 `json:"corruptLines"`
|
||||
Fallbacks uint64 `json:"fallbacks"`
|
||||
LastError string `json:"lastError,omitempty"`
|
||||
}
|
||||
|
||||
type Catalog struct {
|
||||
db *sql.DB
|
||||
statusMu sync.RWMutex
|
||||
status Status
|
||||
revision atomic.Uint64
|
||||
fallback atomic.Uint64
|
||||
queue chan receiptEntry
|
||||
dirtyFiles sync.Map
|
||||
dirtyDirs sync.Map
|
||||
dirtyWake chan struct{}
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
wg sync.WaitGroup
|
||||
}
|
||||
|
||||
type receiptEntry struct {
|
||||
receipt AppendReceipt
|
||||
entry Entry
|
||||
flush chan struct{}
|
||||
}
|
||||
|
||||
func DefaultPath() string { return filepath.Join(config.CacheDir(), "usage-catalog", "v1.sqlite") }
|
||||
|
||||
const schema = `
|
||||
CREATE TABLE usage_state(id INTEGER PRIMARY KEY CHECK(id=1),revision INTEGER NOT NULL DEFAULT 0);
|
||||
INSERT INTO usage_state(id,revision) VALUES(1,0);
|
||||
CREATE TABLE usage_files(
|
||||
path TEXT PRIMARY KEY,day TEXT NOT NULL,size INTEGER NOT NULL DEFAULT 0,mtime_ns INTEGER NOT NULL DEFAULT 0,
|
||||
indexed_offset INTEGER NOT NULL DEFAULT 0,state TEXT NOT NULL DEFAULT 'pending',error TEXT NOT NULL DEFAULT '',
|
||||
corrupt_lines INTEGER NOT NULL DEFAULT 0,completed_at INTEGER NOT NULL DEFAULT 0
|
||||
);
|
||||
CREATE TABLE usage_records(
|
||||
file_path TEXT NOT NULL,byte_offset INTEGER NOT NULL,byte_length INTEGER NOT NULL,line_hash TEXT NOT NULL,
|
||||
day TEXT NOT NULL,source TEXT NOT NULL,model_ref TEXT NOT NULL,provider TEXT NOT NULL,
|
||||
prompt INTEGER NOT NULL,completion INTEGER NOT NULL,reasoning INTEGER NOT NULL,cache_hit INTEGER NOT NULL,
|
||||
cache_miss INTEGER NOT NULL,total INTEGER NOT NULL,requests INTEGER NOT NULL,turns INTEGER NOT NULL,
|
||||
PRIMARY KEY(file_path,byte_offset)
|
||||
);
|
||||
CREATE TABLE usage_rollups(
|
||||
day TEXT NOT NULL,source TEXT NOT NULL,model_ref TEXT NOT NULL,provider TEXT NOT NULL,
|
||||
prompt INTEGER NOT NULL,completion INTEGER NOT NULL,reasoning INTEGER NOT NULL,cache_hit INTEGER NOT NULL,
|
||||
cache_miss INTEGER NOT NULL,total INTEGER NOT NULL,requests INTEGER NOT NULL,turns INTEGER NOT NULL,
|
||||
PRIMARY KEY(day,source,model_ref)
|
||||
);
|
||||
CREATE INDEX idx_usage_rollups_range ON usage_rollups(day,source,model_ref);
|
||||
`
|
||||
|
||||
func migrations() []projectiondb.Migration {
|
||||
return []projectiondb.Migration{{Version: 1, Apply: func(ctx context.Context, tx *sql.Tx) error {
|
||||
_, err := tx.ExecContext(ctx, schema)
|
||||
return err
|
||||
}}}
|
||||
}
|
||||
|
||||
func Open(ctx context.Context, path string) (*Catalog, error) {
|
||||
if path == "" {
|
||||
path = DefaultPath()
|
||||
}
|
||||
handle, err := projectiondb.Open(ctx, projectiondb.OpenOptions{Path: path, MemoryName: "usage-catalog", Migrations: migrations(), MaxOpenConns: 4})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
workerCtx, cancel := context.WithCancel(context.Background())
|
||||
c := &Catalog{db: handle.DB, queue: make(chan receiptEntry, 1024), dirtyWake: make(chan struct{}, 1), ctx: workerCtx, cancel: cancel,
|
||||
status: Status{State: string(handle.Status.State), Mode: handle.Status.Mode, Path: handle.Status.Path, LastError: handle.Status.LastError}}
|
||||
var revision uint64
|
||||
_ = c.db.QueryRowContext(ctx, `SELECT revision FROM usage_state WHERE id=1`).Scan(&revision)
|
||||
c.revision.Store(revision)
|
||||
c.refresh(ctx)
|
||||
c.wg.Add(1)
|
||||
go c.worker()
|
||||
return c, nil
|
||||
}
|
||||
|
||||
func (c *Catalog) Enqueue(receipt AppendReceipt, entry Entry) bool {
|
||||
if c == nil {
|
||||
return false
|
||||
}
|
||||
select {
|
||||
case c.queue <- receiptEntry{receipt: receipt, entry: entry}:
|
||||
return true
|
||||
default:
|
||||
c.dirtyFiles.Store(receipt.Path, receipt.Day)
|
||||
c.wakeDirty()
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Catalog) worker() {
|
||||
defer c.wg.Done()
|
||||
for {
|
||||
if path, day, ok := c.takeDirtyFile(); ok {
|
||||
_ = c.ReconcileFile(c.ctx, path, day)
|
||||
continue
|
||||
}
|
||||
if dir, ok := c.takeDirtyDir(); ok {
|
||||
_ = c.ReconcileDir(c.ctx, dir)
|
||||
continue
|
||||
}
|
||||
select {
|
||||
case <-c.ctx.Done():
|
||||
return
|
||||
case <-c.dirtyWake:
|
||||
case item := <-c.queue:
|
||||
if item.flush != nil {
|
||||
close(item.flush)
|
||||
continue
|
||||
}
|
||||
_ = c.applyReceipt(c.ctx, item.receipt, item.entry)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Catalog) takeDirtyFile() (string, string, bool) {
|
||||
var path, day string
|
||||
c.dirtyFiles.Range(func(key, value any) bool {
|
||||
path, _ = key.(string)
|
||||
day, _ = value.(string)
|
||||
c.dirtyFiles.Delete(key)
|
||||
return false
|
||||
})
|
||||
return path, day, path != ""
|
||||
}
|
||||
|
||||
func (c *Catalog) takeDirtyDir() (string, bool) {
|
||||
var dir string
|
||||
c.dirtyDirs.Range(func(key, _ any) bool {
|
||||
dir, _ = key.(string)
|
||||
c.dirtyDirs.Delete(key)
|
||||
return false
|
||||
})
|
||||
return dir, dir != ""
|
||||
}
|
||||
|
||||
func (c *Catalog) RequestReconcileDir(dir string) {
|
||||
if c != nil && strings.TrimSpace(dir) != "" {
|
||||
c.dirtyDirs.Store(filepath.Clean(dir), true)
|
||||
c.wakeDirty()
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Catalog) wakeDirty() {
|
||||
select {
|
||||
case c.dirtyWake <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Catalog) Flush(ctx context.Context) error {
|
||||
if c == nil {
|
||||
return nil
|
||||
}
|
||||
done := make(chan struct{})
|
||||
select {
|
||||
case c.queue <- receiptEntry{flush: done}:
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
select {
|
||||
case <-done:
|
||||
return nil
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Catalog) applyReceipt(ctx context.Context, receipt AppendReceipt, entry Entry) error {
|
||||
tx, err := c.db.BeginTx(ctx, nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var existingHash string
|
||||
err = tx.QueryRowContext(ctx, `SELECT line_hash FROM usage_records WHERE file_path=? AND byte_offset=?`, receipt.Path, receipt.Offset).Scan(&existingHash)
|
||||
if err == nil {
|
||||
_ = tx.Rollback()
|
||||
if existingHash != receipt.LineHash {
|
||||
return c.ReconcileFile(ctx, receipt.Path, receipt.Day)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
if !errors.Is(err, sql.ErrNoRows) {
|
||||
_ = tx.Rollback()
|
||||
return err
|
||||
}
|
||||
if err := insertRecord(ctx, tx, receipt, entry); err != nil {
|
||||
_ = tx.Rollback()
|
||||
return err
|
||||
}
|
||||
end := receipt.Offset + int64(receipt.Length)
|
||||
_, err = tx.ExecContext(ctx, `INSERT INTO usage_files(path,day,size,indexed_offset,state,completed_at) VALUES(?,?,?,?,'ready',?)
|
||||
ON CONFLICT(path) DO UPDATE SET day=excluded.day,size=MAX(usage_files.size,excluded.size),
|
||||
indexed_offset=CASE WHEN usage_files.indexed_offset=? THEN excluded.indexed_offset ELSE usage_files.indexed_offset END,
|
||||
state=CASE WHEN usage_files.indexed_offset=? THEN 'ready' ELSE 'pending' END,completed_at=excluded.completed_at`,
|
||||
receipt.Path, receipt.Day, end, end, time.Now().UnixMilli(), receipt.Offset, receipt.Offset)
|
||||
if err != nil {
|
||||
_ = tx.Rollback()
|
||||
return err
|
||||
}
|
||||
revision, err := bump(ctx, tx)
|
||||
if err != nil {
|
||||
_ = tx.Rollback()
|
||||
return err
|
||||
}
|
||||
if err := tx.Commit(); err != nil {
|
||||
return err
|
||||
}
|
||||
c.revision.Store(revision)
|
||||
c.refresh(context.Background())
|
||||
return nil
|
||||
}
|
||||
|
||||
func insertRecord(ctx context.Context, tx *sql.Tx, receipt AppendReceipt, entry Entry) error {
|
||||
if entry.Total > 0 && entry.Requests <= 0 {
|
||||
entry.Requests = 1
|
||||
}
|
||||
result, err := tx.ExecContext(ctx, `INSERT OR IGNORE INTO usage_records(file_path,byte_offset,byte_length,line_hash,day,source,
|
||||
model_ref,provider,prompt,completion,reasoning,cache_hit,cache_miss,total,requests,turns) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`,
|
||||
receipt.Path, receipt.Offset, receipt.Length, receipt.LineHash, entry.Day, entry.Source, entry.ModelRef, entry.Provider,
|
||||
entry.Prompt, entry.Completion, entry.Reasoning, entry.CacheHit, entry.CacheMiss, entry.Total, entry.Requests, entry.Turns)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
inserted, _ := result.RowsAffected()
|
||||
if inserted == 0 {
|
||||
return nil
|
||||
}
|
||||
_, err = tx.ExecContext(ctx, `INSERT INTO usage_rollups(day,source,model_ref,provider,prompt,completion,reasoning,cache_hit,
|
||||
cache_miss,total,requests,turns) VALUES(?,?,?,?,?,?,?,?,?,?,?,?) ON CONFLICT(day,source,model_ref) DO UPDATE SET
|
||||
prompt=prompt+excluded.prompt,completion=completion+excluded.completion,reasoning=reasoning+excluded.reasoning,
|
||||
cache_hit=cache_hit+excluded.cache_hit,cache_miss=cache_miss+excluded.cache_miss,total=total+excluded.total,
|
||||
requests=requests+excluded.requests,turns=turns+excluded.turns`, entry.Day, entry.Source, entry.ModelRef, entry.Provider,
|
||||
entry.Prompt, entry.Completion, entry.Reasoning, entry.CacheHit, entry.CacheMiss, entry.Total, entry.Requests, entry.Turns)
|
||||
return err
|
||||
}
|
||||
|
||||
type rawRecord struct {
|
||||
Timestamp time.Time `json:"ts"`
|
||||
ModelRef string `json:"model"`
|
||||
Source string `json:"source"`
|
||||
Prompt int `json:"prompt"`
|
||||
Completion int `json:"completion"`
|
||||
Reasoning int `json:"reasoning"`
|
||||
CacheHit int `json:"cache_hit"`
|
||||
CacheMiss int `json:"cache_miss"`
|
||||
Total int `json:"total"`
|
||||
Requests int `json:"requests"`
|
||||
Turn bool `json:"turn"`
|
||||
}
|
||||
|
||||
func providerOf(model string) string {
|
||||
if i := strings.IndexByte(model, '/'); i > 0 {
|
||||
return model[:i]
|
||||
}
|
||||
return "default"
|
||||
}
|
||||
|
||||
func entryFromRaw(day string, raw rawRecord) Entry {
|
||||
turns := 0
|
||||
if raw.Turn {
|
||||
turns = 1
|
||||
}
|
||||
return Entry{Day: day, Source: raw.Source, ModelRef: raw.ModelRef, Provider: providerOf(raw.ModelRef), Prompt: raw.Prompt,
|
||||
Completion: raw.Completion, Reasoning: raw.Reasoning, CacheHit: raw.CacheHit, CacheMiss: raw.CacheMiss,
|
||||
Total: raw.Total, Requests: raw.Requests, Turns: turns}
|
||||
}
|
||||
|
||||
func (c *Catalog) ReconcileFile(ctx context.Context, path, day string) error {
|
||||
f, err := os.Open(path)
|
||||
if errors.Is(err, os.ErrNotExist) {
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer f.Close()
|
||||
tx, err := c.db.BeginTx(ctx, nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := tx.ExecContext(ctx, `DELETE FROM usage_rollups WHERE day IN (SELECT DISTINCT day FROM usage_records WHERE file_path=?)`, path); err != nil {
|
||||
_ = tx.Rollback()
|
||||
return err
|
||||
}
|
||||
if _, err := tx.ExecContext(ctx, `DELETE FROM usage_records WHERE file_path=?`, path); err != nil {
|
||||
_ = tx.Rollback()
|
||||
return err
|
||||
}
|
||||
reader := bufio.NewReader(f)
|
||||
offset := int64(0)
|
||||
corrupt := int64(0)
|
||||
for {
|
||||
line, readErr := reader.ReadBytes('\n')
|
||||
if len(line) > 0 {
|
||||
trimmed := strings.TrimSpace(string(line))
|
||||
if trimmed != "" {
|
||||
var raw rawRecord
|
||||
if json.Unmarshal([]byte(trimmed), &raw) != nil {
|
||||
corrupt++
|
||||
} else {
|
||||
hash := sha256.Sum256([]byte(trimmed))
|
||||
receipt := AppendReceipt{Path: path, Day: day, Offset: offset, Length: len(line), LineHash: hex.EncodeToString(hash[:])}
|
||||
if err := insertRecord(ctx, tx, receipt, entryFromRaw(day, raw)); err != nil {
|
||||
_ = tx.Rollback()
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
offset += int64(len(line))
|
||||
}
|
||||
if errors.Is(readErr, io.EOF) {
|
||||
break
|
||||
}
|
||||
if readErr != nil {
|
||||
_ = tx.Rollback()
|
||||
return readErr
|
||||
}
|
||||
}
|
||||
info, _ := os.Stat(path)
|
||||
mtime := int64(0)
|
||||
if info != nil {
|
||||
mtime = info.ModTime().UnixNano()
|
||||
}
|
||||
_, err = tx.ExecContext(ctx, `INSERT INTO usage_files(path,day,size,mtime_ns,indexed_offset,state,error,corrupt_lines,completed_at)
|
||||
VALUES(?,?,?,?,?,'ready','',?,?) ON CONFLICT(path) DO UPDATE SET day=excluded.day,size=excluded.size,mtime_ns=excluded.mtime_ns,
|
||||
indexed_offset=excluded.indexed_offset,state='ready',error='',corrupt_lines=excluded.corrupt_lines,completed_at=excluded.completed_at`,
|
||||
path, day, offset, mtime, offset, corrupt, time.Now().UnixMilli())
|
||||
if err != nil {
|
||||
_ = tx.Rollback()
|
||||
return err
|
||||
}
|
||||
revision, err := bump(ctx, tx)
|
||||
if err != nil {
|
||||
_ = tx.Rollback()
|
||||
return err
|
||||
}
|
||||
if err := tx.Commit(); err != nil {
|
||||
return err
|
||||
}
|
||||
c.revision.Store(revision)
|
||||
c.refresh(context.Background())
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Catalog) ReconcileDir(ctx context.Context, dir string) error {
|
||||
entries, err := os.ReadDir(dir)
|
||||
if errors.Is(err, os.ErrNotExist) {
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, entry := range entries {
|
||||
if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".jsonl") {
|
||||
continue
|
||||
}
|
||||
day := strings.TrimSuffix(entry.Name(), ".jsonl")
|
||||
if err := c.ReconcileFile(ctx, filepath.Join(dir, entry.Name()), day); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Catalog) Ready(ctx context.Context, dir string, days []string) bool {
|
||||
for _, day := range days {
|
||||
path := filepath.Join(dir, day+".jsonl")
|
||||
info, err := os.Stat(path)
|
||||
if errors.Is(err, os.ErrNotExist) {
|
||||
continue
|
||||
}
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
var size, offset int64
|
||||
var state string
|
||||
if err := c.db.QueryRowContext(ctx, `SELECT size,indexed_offset,state FROM usage_files WHERE path=?`, path).Scan(&size, &offset, &state); err != nil {
|
||||
return false
|
||||
}
|
||||
if state != "ready" || size != info.Size() || offset != info.Size() {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func (c *Catalog) Query(ctx context.Context, fromDay, toDay, source string) ([]Rollup, error) {
|
||||
args := []any{fromDay, toDay}
|
||||
where := `day>=? AND day<=?`
|
||||
if source != "" && source != "all" {
|
||||
where += ` AND source=?`
|
||||
args = append(args, source)
|
||||
}
|
||||
rows, err := c.db.QueryContext(ctx, `SELECT day,source,model_ref,provider,prompt,completion,reasoning,cache_hit,cache_miss,total,requests,turns
|
||||
FROM usage_rollups WHERE `+where+` ORDER BY day,source,model_ref`, args...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
out := []Rollup{}
|
||||
for rows.Next() {
|
||||
var row Rollup
|
||||
if err := rows.Scan(&row.Day, &row.Source, &row.ModelRef, &row.Provider, &row.Prompt, &row.Completion, &row.Reasoning,
|
||||
&row.CacheHit, &row.CacheMiss, &row.Total, &row.Requests, &row.Turns); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, row)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
func bump(ctx context.Context, tx *sql.Tx) (uint64, error) {
|
||||
if _, err := tx.ExecContext(ctx, `UPDATE usage_state SET revision=revision+1 WHERE id=1`); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
var revision uint64
|
||||
err := tx.QueryRowContext(ctx, `SELECT revision FROM usage_state WHERE id=1`).Scan(&revision)
|
||||
return revision, err
|
||||
}
|
||||
|
||||
func (c *Catalog) NoteFallback() { c.fallback.Add(1) }
|
||||
|
||||
func (c *Catalog) refresh(ctx context.Context) {
|
||||
var files, lag, corrupt int64
|
||||
_ = c.db.QueryRowContext(ctx, `SELECT COUNT(*),COALESCE(SUM(MAX(size-indexed_offset,0)),0),COALESCE(SUM(corrupt_lines),0) FROM usage_files`).Scan(&files, &lag, &corrupt)
|
||||
c.statusMu.Lock()
|
||||
c.status.Revision, c.status.IndexedFiles, c.status.LagBytes, c.status.CorruptLines = c.revision.Load(), files, lag, corrupt
|
||||
c.status.Fallbacks = c.fallback.Load()
|
||||
c.statusMu.Unlock()
|
||||
}
|
||||
|
||||
func (c *Catalog) Status() Status {
|
||||
c.statusMu.RLock()
|
||||
defer c.statusMu.RUnlock()
|
||||
status := c.status
|
||||
status.Fallbacks = c.fallback.Load()
|
||||
return status
|
||||
}
|
||||
|
||||
func (c *Catalog) Close(ctx context.Context) error {
|
||||
c.cancel()
|
||||
done := make(chan struct{})
|
||||
go func() { c.wg.Wait(); close(done) }()
|
||||
select {
|
||||
case <-done:
|
||||
return c.db.Close()
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,74 @@
|
||||
package usagecatalog
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestReconcileFileAndDuplicateReceiptAreIdempotent(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
dir := t.TempDir()
|
||||
path := filepath.Join(dir, "2026-08-10.jsonl")
|
||||
line := []byte("{\"ts\":\"2026-08-10T10:00:00+08:00\",\"model\":\"deepseek/model\",\"source\":\"desktop\",\"total\":42}\n")
|
||||
if err := os.WriteFile(path, line, 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
catalog, err := Open(ctx, filepath.Join(t.TempDir(), "usage.sqlite"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = catalog.Close(context.Background()) })
|
||||
if err := catalog.ReconcileFile(ctx, path, "2026-08-10"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
rows, err := catalog.Query(ctx, "2026-08-10", "2026-08-10", "desktop")
|
||||
if err != nil || len(rows) != 1 || rows[0].Total != 42 || rows[0].Requests != 1 {
|
||||
t.Fatalf("rows=%#v err=%v", rows, err)
|
||||
}
|
||||
var receipt AppendReceipt
|
||||
if err := catalog.db.QueryRow(`SELECT file_path,day,byte_offset,byte_length,line_hash FROM usage_records WHERE file_path=?`, path).Scan(
|
||||
&receipt.Path, &receipt.Day, &receipt.Offset, &receipt.Length, &receipt.LineHash); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
entry := Entry{Day: receipt.Day, Source: "desktop", ModelRef: "deepseek/model", Provider: "deepseek", Total: 42, Requests: 1}
|
||||
if err := catalog.applyReceipt(ctx, receipt, entry); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
rows, err = catalog.Query(ctx, "2026-08-10", "2026-08-10", "desktop")
|
||||
if err != nil || len(rows) != 1 || rows[0].Total != 42 {
|
||||
t.Fatalf("duplicate changed aggregate: rows=%#v err=%v", rows, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestReadyRejectsExternalAppendUntilReconciled(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
dir := t.TempDir()
|
||||
path := filepath.Join(dir, "2026-08-10.jsonl")
|
||||
if err := os.WriteFile(path, []byte("{\"ts\":\"2026-08-10T10:00:00Z\",\"total\":1}\n"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
catalog, err := Open(ctx, filepath.Join(t.TempDir(), "usage.sqlite"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = catalog.Close(context.Background()) })
|
||||
if err := catalog.ReconcileFile(ctx, path, "2026-08-10"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !catalog.Ready(ctx, dir, []string{"2026-08-10"}) {
|
||||
t.Fatal("catalog should be ready")
|
||||
}
|
||||
f, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0o600)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_, _ = f.WriteString("{\"ts\":\"2026-08-10T11:00:00Z\",\"total\":2}\n")
|
||||
_ = f.Close()
|
||||
if catalog.Ready(ctx, dir, []string{"2026-08-10"}) {
|
||||
t.Fatal("external append was treated as indexed")
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user