bucketeer/internal/store/state.go

package store

import (
	"context"
	"crypto/sha256"
	"encoding/json"
	"fmt"
	"log"
	"strconv"
	"time"

	"github.com/jackc/pgx/v5"

	"software.sslmate.com/src/certspotter/cttypes"
	"software.sslmate.com/src/certspotter/loglist"
	"software.sslmate.com/src/certspotter/monitor"
)

var _ monitor.StateProvider = (*Store)(nil)

// Prepare implements monitor.StateProvider. It applies the embedded schema so
// the container needs no writable filesystem.
func (s *Store) Prepare(ctx context.Context) error {
	if err := s.Migrate(ctx); err != nil {
		return err
	}
	return nil
}

// PrepareLog is a no-op: checkpoint rows are created lazily by StoreLogState.
func (s *Store) PrepareLog(ctx context.Context, logID monitor.LogID) error { return nil }

// StoreLogState persists the certspotter monitor position for a log.
func (s *Store) StoreLogState(ctx context.Context, logID monitor.LogID, state *monitor.LogState) error {
	data, err := json.Marshal(state)
	if err != nil {
		return err
	}
	key := logID.Base64URLString()
	hash := sha256.Sum256(data)
	s.stateMu.Lock()
	if prev, ok := s.lastState[key]; ok && prev == hash {
		s.stateMu.Unlock()
		return nil // unchanged; CertSpotter calls this per batch
	}
	s.lastState[key] = hash
	s.stateMu.Unlock()

	_, err = s.pool.Exec(ctx, `
		INSERT INTO ct_checkpoints (log_id, state, updated_at)
		VALUES ($1, $2, now())
		ON CONFLICT (log_id) DO UPDATE SET state = EXCLUDED.state, updated_at = now()`,
		key, data)
	return err
}

// LoadLogState returns nil when no checkpoint exists for the log.
func (s *Store) LoadLogState(ctx context.Context, logID monitor.LogID) (*monitor.LogState, error) {
	var data []byte
	err := s.pool.QueryRow(ctx, `SELECT state FROM ct_checkpoints WHERE log_id = $1`, logID.Base64URLString()).Scan(&data)
	if err == pgx.ErrNoRows {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	state := new(monitor.LogState)
	if err := json.Unmarshal(data, state); err != nil {
		return nil, fmt.Errorf("corrupt checkpoint for log %s: %w", logID.Base64URLString(), err)
	}
	return state, nil
}

func sthKey(sth *cttypes.SignedTreeHead) string {
	return strconv.FormatUint(sth.TreeSize, 10) + "-" +
		strconv.FormatUint(sth.Timestamp, 10) + "-" +
		sth.RootHash.Base64String()
}

// StoreSTH stores an unverified signed tree head. If it already exists the
// original stored_at is preserved.
func (s *Store) StoreSTH(ctx context.Context, logID monitor.LogID, sth *cttypes.SignedTreeHead) (*monitor.StoredSTH, error) {
	data, err := json.Marshal(sth)
	if err != nil {
		return nil, err
	}
	var storedAt time.Time
	err = s.pool.QueryRow(ctx, `
		INSERT INTO ct_sths (log_id, sth_key, tree_size, sth, stored_at)
		VALUES ($1, $2, $3, $4, now())
		ON CONFLICT (log_id, sth_key) DO UPDATE SET log_id = EXCLUDED.log_id
		RETURNING stored_at`,
		logID.Base64URLString(), sthKey(sth), sth.TreeSize, data).Scan(&storedAt)
	if err != nil {
		return nil, err
	}
	return &monitor.StoredSTH{SignedTreeHead: *sth, StoredAt: storedAt}, nil
}

// LoadSTHs loads all stored STHs for a log, sorted by tree size.
func (s *Store) LoadSTHs(ctx context.Context, logID monitor.LogID) ([]*monitor.StoredSTH, error) {
	rows, err := s.pool.Query(ctx,
		`SELECT sth, stored_at FROM ct_sths WHERE log_id = $1 ORDER BY tree_size ASC`,
		logID.Base64URLString())
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []*monitor.StoredSTH
	for rows.Next() {
		var (
			data     []byte
			storedAt time.Time
		)
		if err := rows.Scan(&data, &storedAt); err != nil {
			return nil, err
		}
		var sth cttypes.SignedTreeHead
		if err := json.Unmarshal(data, &sth); err != nil {
			return nil, err
		}
		out = append(out, &monitor.StoredSTH{SignedTreeHead: sth, StoredAt: storedAt})
	}
	return out, rows.Err()
}

// RemoveSTH deletes an STH once it has been verified against the Merkle tree.
func (s *Store) RemoveSTH(ctx context.Context, logID monitor.LogID, sth *cttypes.SignedTreeHead) error {
	_, err := s.pool.Exec(ctx, `DELETE FROM ct_sths WHERE log_id = $1 AND sth_key = $2`,
		logID.Base64URLString(), sthKey(sth))
	return err
}

// StoreIssuer and LoadIssuer are no-ops. bucketeer's identifier sink bypasses
// certificate-chain retrieval entirely, so the issuer cache is never consulted;
// keeping it empty also satisfies the "no certificate contents in the state
// store" requirement. The Go runtime cache handles any incidental calls.
func (s *Store) StoreIssuer(ctx context.Context, fingerprint *[32]byte, issuer []byte) error {
	return nil
}

func (s *Store) LoadIssuer(ctx context.Context, fingerprint *[32]byte) ([]byte, error) {
	return nil, nil
}

// NotifyCert is never called because the identifier sink handles certificates.
func (s *Store) NotifyCert(ctx context.Context, cert *monitor.DiscoveredCert) error { return nil }

// NotifyMalformedEntry counts unparseable log entries without logging per entry.
func (s *Store) NotifyMalformedEntry(ctx context.Context, entry *monitor.LogEntry, parseError error) error {
	if s.metrics != nil {
		s.metrics.MalformedCT.Add(1)
	}
	return nil
}

// NotifyHealthCheckFailure is rare (once per health-check interval at most), so
// it is safe to log.
func (s *Store) NotifyHealthCheckFailure(ctx context.Context, ctlog *loglist.Log, info monitor.HealthCheckFailure) error {
	log.Printf("ct health check failure: %s", info.Summary())
	return nil
}

// NotifyError counts transient CT errors. It deliberately does not log, to keep
// logging off the high-frequency error path.
func (s *Store) NotifyError(ctx context.Context, ctlog *loglist.Log, notifyErr error) error {
	if s.metrics != nil {
		s.metrics.CTErrors.Add(1)
	}
	return nil
}

// GetErrors returns the recent CT errors. bucketeer keeps only aggregate counters.
func (s *Store) GetErrors(ctx context.Context, ctlog *loglist.Log, count int) (string, error) {
	return "", nil
}