package storage

import (
	"context"
	"fmt"
	"time"
)

const maxAuditPruneBatch = 100000

var afterAuditPruneBatch = func(int64) {}

func (s *Store) PruneAudit(ctx context.Context, before time.Time, batch int, pause time.Duration) (int64, error) {
	const operation = "prune audit"
	if err := s.checkOpen(operation); err != nil {
		return 0, err
	}
	if batch < 1 || batch > maxAuditPruneBatch {
		return 0, invalid(operation, fmt.Errorf("batch must be between 1 and %d", maxAuditPruneBatch))
	}
	var total int64
	for {
		n, err := s.pruneAuditBatch(ctx, timeToInt(before), batch)
		total += n
		if err != nil {
			return total, err
		}
		afterAuditPruneBatch(n)
		if n < int64(batch) {
			return total, nil
		}
		if pause > 0 {
			t := time.NewTimer(pause)
			select {
			case <-ctx.Done():
				t.Stop()
				return total, ctx.Err()
			case <-t.C:
			}
		} else if err := ctx.Err(); err != nil {
			return total, err
		}
	}
}

func (s *Store) pruneAuditBatch(ctx context.Context, before int64, batch int) (int64, error) {
	const operation = "prune audit batch"
	tx, err := s.sql.BeginTx(ctx, nil)
	if err != nil {
		return 0, classify("begin "+operation, err)
	}
	defer tx.Rollback()
	result, err := tx.ExecContext(ctx, `
DELETE FROM audit_log WHERE id IN (
    SELECT id FROM audit_log WHERE occurred_at < ? ORDER BY occurred_at, id LIMIT ?
)`, before, batch)
	if err != nil {
		return 0, classify(operation, err)
	}
	n, err := result.RowsAffected()
	if err != nil {
		return 0, classify(operation, err)
	}
	if err := tx.Commit(); err != nil {
		return 0, classify("commit "+operation, err)
	}
	return n, nil
}