nuntius/internal/storage/pending.go

// Copyright (c) 2026 Petr Balvín <opensource@petrbalvin.org> (https://petrbalvin.org)
// SPDX-License-Identifier: MIT

//go:build linux || freebsd

// Double opt-in support: addresses wait in a pending file until their
// confirmation token is redeemed, then move into the main subscriber log.

package storage

import (
	"crypto/rand"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"os"
	"path/filepath"
	"sync"
	"time"
)

// PendingTTL is how long an unconfirmed subscription stays actionable.
// After that it is purged and the address must be signed up again.
const PendingTTL = 72 * time.Hour

// RandomToken returns a 32-byte cryptographically random value hex-encoded
// for use inside URLs. Only its SHA-256 hash is persisted; the raw value
// lives exclusively in the confirmation link.
func RandomToken() (string, error) {
	var buf [32]byte
	if _, err := rand.Read(buf[:]); err != nil {
		return "", fmt.Errorf("generate confirmation token: %w", err)
	}
	return hex.EncodeToString(buf[:]), nil
}

type PendingSubscription struct {
	Email     string    `json:"email"`
	IP        string    `json:"ip,omitempty"`
	CreatedAt time.Time `json:"created_at"`
}

type pendingFile struct {
	Schema  int                            `json:"schema"`
	Entries map[string]PendingSubscription `json:"entries"`
}

const pendingSchema = 1

// PendingStore keeps unconfirmed newsletter subscriptions keyed by the hash
// of their confirmation token. The whole set is rewritten atomically on
// every change: pending files stay tiny (only signups within the TTL), so
// the append-only trick is not needed here.
type PendingStore struct {
	mu   sync.Mutex
	path string
	ttl  time.Duration
}

// NewPendingStore wraps path with the given entry lifetime.
func NewPendingStore(path string, ttl time.Duration) *PendingStore {
	if ttl <= 0 {
		ttl = PendingTTL
	}
	return &PendingStore{path: path, ttl: ttl}
}

// Path returns the backing file location.
func (p *PendingStore) Path() string { return p.path }

// TTL returns the entry lifetime the store enforces.
func (p *PendingStore) TTL() time.Duration { return p.ttl }

func hashToken(raw string) string {
	sum := sha256.Sum256([]byte(raw))
	return hex.EncodeToString(sum[:])
}

// load reads the file, prunes expired entries and persists the pruned set.
// A missing or corrupt file behaves like an empty store.
func (p *PendingStore) load(now time.Time) (map[string]PendingSubscription, error) {
	f := pendingFile{Schema: pendingSchema, Entries: map[string]PendingSubscription{}}
	raw, err := os.ReadFile(p.path)
	switch {
	case err == nil:
		if err := json.Unmarshal(raw, &f); err != nil || f.Schema != pendingSchema {
			return f.Entries, fmt.Errorf("unreadable pending store %s", p.path)
		}
	case os.IsNotExist(err):
	default:
		return f.Entries, fmt.Errorf("read pending store %s: %w", p.path, err)
	}
	dirty := false
	for k, e := range f.Entries {
		if now.Sub(e.CreatedAt) > p.ttl || e.CreatedAt.After(now.Add(time.Hour)) {
			delete(f.Entries, k)
			dirty = true
		}
	}
	if dirty {
		if werr := p.save(f); werr != nil {
			return f.Entries, werr
		}
	}
	return f.Entries, nil
}

func (p *PendingStore) save(f pendingFile) error {
	line, err := json.Marshal(f)
	if err != nil {
		return fmt.Errorf("marshal pending store: %w", err)
	}
	if err := os.MkdirAll(filepath.Dir(p.path), 0o755); err != nil {
		return fmt.Errorf("mkdir %s: %w", filepath.Dir(p.path), err)
	}
	tmp := p.path + ".tmp"
	if err := os.WriteFile(tmp, line, 0o600); err != nil {
		return fmt.Errorf("write %s: %w", tmp, err)
	}
	return os.Rename(tmp, p.path)
}

// Issue stores a new pending subscription keyed by the token hash,
// superseding any earlier entry for the same address. A load warning
// (unreadable file) is non-fatal: the new entry is still written.
func (p *PendingStore) Issue(rawToken string, sub PendingSubscription) error {
	p.mu.Lock()
	defer p.mu.Unlock()
	now := time.Now()
	entries, _ := p.load(now)
	for k, e := range entries {
		if seenKey(e.Email) == seenKey(sub.Email) && k != hashToken(rawToken) {
			delete(entries, k) // one live token per address
		}
	}
	sub.CreatedAt = now.UTC()
	entries[hashToken(rawToken)] = sub
	return p.save(pendingFile{Schema: pendingSchema, Entries: entries})
}

// Consume redeems a token: a valid, unexpired entry is removed from the
// file and returned. Unknown tokens, already-redeemed tokens and expired
// entries all report false.
func (p *PendingStore) Consume(rawToken string) (PendingSubscription, bool) {
	p.mu.Lock()
	defer p.mu.Unlock()
	key := hashToken(rawToken)
	now := time.Now()
	entries, _ := p.load(now)
	sub, ok := entries[key]
	if !ok {
		return PendingSubscription{}, false
	}
	delete(entries, key)
	_ = p.save(pendingFile{Schema: pendingSchema, Entries: entries})
	if now.Sub(sub.CreatedAt) > p.ttl {
		return PendingSubscription{}, false
	}
	return sub, true
}

// Peek returns the subscription behind rawToken without consuming it,
// reporting false when the token is unknown or expired.
func (p *PendingStore) Peek(rawToken string) (PendingSubscription, bool) {
	p.mu.Lock()
	defer p.mu.Unlock()
	now := time.Now()
	entries, _ := p.load(now)
	sub, ok := entries[hashToken(rawToken)]
	if !ok || now.Sub(sub.CreatedAt) > p.ttl {
		return PendingSubscription{}, false
	}
	return sub, true
}

// HasToken reports whether raw is still a live, redeemable token.
func (p *PendingStore) HasToken(rawToken string) bool {
	p.mu.Lock()
	defer p.mu.Unlock()
	now := time.Now()
	entries, _ := p.load(now)
	sub, ok := entries[hashToken(rawToken)]
	return ok && now.Sub(sub.CreatedAt) <= p.ttl
}