//go:build !(js && wasm) package bbolt import ( "context" "errors" "os" "path/filepath" "sync" "time" bolt "go.etcd.io/bbolt" "lol.mleku.dev" "lol.mleku.dev/chk" "next.orly.dev/pkg/database" "next.orly.dev/pkg/utils/apputil" ) // Bucket names - map to existing index prefixes but without the 3-byte prefix in keys var ( bucketEvt = []byte("evt") // Event storage: serial -> compact event data bucketEid = []byte("eid") // Event ID index bucketFpc = []byte("fpc") // Full ID/pubkey index bucketC = []byte("c--") // Created at index bucketKc = []byte("kc-") // Kind + created index bucketPc = []byte("pc-") // Pubkey + created index bucketKpc = []byte("kpc") // Kind + pubkey + created bucketTc = []byte("tc-") // Tag + created bucketTkc = []byte("tkc") // Tag + kind + created bucketTpc = []byte("tpc") // Tag + pubkey + created bucketTkp = []byte("tkp") // Tag + kind + pubkey + created bucketWrd = []byte("wrd") // Word search index bucketExp = []byte("exp") // Expiration index bucketPks = []byte("pks") // Pubkey hash -> serial bucketSpk = []byte("spk") // Serial -> pubkey bucketSei = []byte("sei") // Serial -> event ID bucketCmp = []byte("cmp") // Compact event storage bucketEv = []byte("ev") // Event vertices (adjacency list) bucketPv = []byte("pv") // Pubkey vertices (adjacency list) bucketMeta = []byte("_meta") // Markers, version, serial counter, bloom filter ) // All buckets that need to be created on init var allBuckets = [][]byte{ bucketEvt, bucketEid, bucketFpc, bucketC, bucketKc, bucketPc, bucketKpc, bucketTc, bucketTkc, bucketTpc, bucketTkp, bucketWrd, bucketExp, bucketPks, bucketSpk, bucketSei, bucketCmp, bucketEv, bucketPv, bucketMeta, } // B implements the database.Database interface using BBolt as the storage backend. // Optimized for HDD with write batching and adjacency list graph storage. type B struct { ctx context.Context cancel context.CancelFunc dataDir string Logger *Logger db *bolt.DB ready chan struct{} // Write batching batcher *WriteBatcher // Serial management serialMu sync.Mutex nextSerial uint64 nextPubkeySeq uint64 // Edge bloom filter for fast negative lookups edgeBloom *EdgeBloomFilter // Configuration cfg *BboltConfig } // BboltConfig holds bbolt-specific configuration type BboltConfig struct { DataDir string LogLevel string // Batch settings (tuned for 7200rpm HDD) BatchMaxEvents int // Max events before flush (default: 5000) BatchMaxBytes int64 // Max bytes before flush (default: 128MB) BatchFlushTimeout time.Duration // Max time before flush (default: 30s) // Bloom filter settings BloomSizeMB int // Bloom filter size in MB (default: 16) // BBolt settings NoSync bool // Disable fsync for performance (DANGEROUS) InitialMmapSize int // Initial mmap size in bytes } // Ensure B implements Database interface at compile time var _ database.Database = (*B)(nil) // New creates a new BBolt database instance with default configuration. func New( ctx context.Context, cancel context.CancelFunc, dataDir, logLevel string, ) (b *B, err error) { cfg := &BboltConfig{ DataDir: dataDir, LogLevel: logLevel, BatchMaxEvents: 5000, BatchMaxBytes: 128 * 1024 * 1024, // 128MB BatchFlushTimeout: 30 * time.Second, BloomSizeMB: 16, InitialMmapSize: 8 * 1024 * 1024 * 1024, // 8GB } return NewWithConfig(ctx, cancel, cfg) } // NewWithConfig creates a new BBolt database instance with full configuration. func NewWithConfig( ctx context.Context, cancel context.CancelFunc, cfg *BboltConfig, ) (b *B, err error) { // Apply defaults if cfg.BatchMaxEvents <= 0 { cfg.BatchMaxEvents = 5000 } if cfg.BatchMaxBytes <= 0 { cfg.BatchMaxBytes = 128 * 1024 * 1024 } if cfg.BatchFlushTimeout <= 0 { cfg.BatchFlushTimeout = 30 * time.Second } if cfg.BloomSizeMB <= 0 { cfg.BloomSizeMB = 16 } if cfg.InitialMmapSize <= 0 { cfg.InitialMmapSize = 8 * 1024 * 1024 * 1024 } b = &B{ ctx: ctx, cancel: cancel, dataDir: cfg.DataDir, Logger: NewLogger(lol.GetLogLevel(cfg.LogLevel), cfg.DataDir), ready: make(chan struct{}), cfg: cfg, } // Ensure the data directory exists if err = os.MkdirAll(cfg.DataDir, 0755); chk.E(err) { return } if err = apputil.EnsureDir(filepath.Join(cfg.DataDir, "dummy")); chk.E(err) { return } // Open BBolt database dbPath := filepath.Join(cfg.DataDir, "orly.db") opts := &bolt.Options{ Timeout: 10 * time.Second, NoSync: cfg.NoSync, InitialMmapSize: cfg.InitialMmapSize, } if b.db, err = bolt.Open(dbPath, 0600, opts); chk.E(err) { return } // Create all buckets if err = b.db.Update(func(tx *bolt.Tx) error { for _, bucket := range allBuckets { if _, err := tx.CreateBucketIfNotExists(bucket); err != nil { return err } } return nil }); chk.E(err) { return } // Initialize serial counters if err = b.initSerialCounters(); chk.E(err) { return } // Initialize bloom filter b.edgeBloom, err = NewEdgeBloomFilter(cfg.BloomSizeMB, b.db) if chk.E(err) { return } // Initialize write batcher b.batcher = NewWriteBatcher(b.db, b.edgeBloom, cfg, b.Logger) // Run migrations b.RunMigrations() // Start warmup and mark ready go b.warmup() // Start background maintenance go b.backgroundLoop() return } // Path returns the path where the database files are stored. func (b *B) Path() string { return b.dataDir } // Init initializes the database with the given path. func (b *B) Init(path string) error { b.dataDir = path return nil } // Sync flushes the database buffers to disk. func (b *B) Sync() error { // Flush pending writes if err := b.batcher.Flush(); err != nil { return err } // Persist bloom filter if err := b.edgeBloom.Persist(b.db); err != nil { return err } // Persist serial counters if err := b.persistSerialCounters(); err != nil { return err } // Sync BBolt return b.db.Sync() } // Close releases resources and closes the database. func (b *B) Close() (err error) { b.Logger.Infof("bbolt: closing database...") // Stop accepting new writes and flush pending if b.batcher != nil { if err = b.batcher.Shutdown(); chk.E(err) { // Log but continue cleanup } } // Persist bloom filter if b.edgeBloom != nil { if err = b.edgeBloom.Persist(b.db); chk.E(err) { // Log but continue cleanup } } // Persist serial counters if err = b.persistSerialCounters(); chk.E(err) { // Log but continue cleanup } // Close BBolt database if b.db != nil { if err = b.db.Close(); chk.E(err) { return } } b.Logger.Infof("bbolt: database closed") return } // Wipe deletes all data in the database. func (b *B) Wipe() error { return b.db.Update(func(tx *bolt.Tx) error { for _, bucket := range allBuckets { if err := tx.DeleteBucket(bucket); err != nil && !errors.Is(err, bolt.ErrBucketNotFound) { return err } if _, err := tx.CreateBucket(bucket); err != nil { return err } } // Reset serial counters b.serialMu.Lock() b.nextSerial = 1 b.nextPubkeySeq = 1 b.serialMu.Unlock() // Reset bloom filter b.edgeBloom.Reset() return nil }) } // SetLogLevel changes the logging level. func (b *B) SetLogLevel(level string) { b.Logger.SetLogLevel(lol.GetLogLevel(level)) } // Ready returns a channel that closes when the database is ready to serve requests. func (b *B) Ready() <-chan struct{} { return b.ready } // warmup performs database warmup operations and closes the ready channel when complete. func (b *B) warmup() { defer close(b.ready) // Give the database time to settle time.Sleep(1 * time.Second) b.Logger.Infof("bbolt: database warmup complete, ready to serve requests") } // backgroundLoop runs periodic maintenance tasks. func (b *B) backgroundLoop() { expirationTicker := time.NewTicker(10 * time.Minute) bloomPersistTicker := time.NewTicker(5 * time.Minute) defer expirationTicker.Stop() defer bloomPersistTicker.Stop() for { select { case <-expirationTicker.C: b.DeleteExpired() case <-bloomPersistTicker.C: if err := b.edgeBloom.Persist(b.db); chk.E(err) { b.Logger.Warningf("bbolt: failed to persist bloom filter: %v", err) } case <-b.ctx.Done(): b.cancel() return } } }