You've already forked agentic-coding-workflow
feat(jobs): mcp_tokens_purge runner (spec §6.5)
- Periodic delete of mcp_tokens where (expired OR revoked) before retention - 24h interval default, 7d retention default - Mirrors acp_events_purge pattern: idempotent, error-tolerant Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,41 @@
|
||||
package runners
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"time"
|
||||
)
|
||||
|
||||
// MCPTokensPurgeRepo is the minimal repo surface this runner needs.
|
||||
// internal/mcp.Repository satisfies it via an adapter in app.go.
|
||||
type MCPTokensPurgeRepo interface {
|
||||
PurgeExpired(ctx context.Context, before time.Time) (int64, error)
|
||||
}
|
||||
|
||||
// MCPTokensPurge deletes mcp_tokens older than retention. Periodic; idempotent.
|
||||
type MCPTokensPurge struct {
|
||||
repo MCPTokensPurgeRepo
|
||||
retention time.Duration
|
||||
log *slog.Logger
|
||||
}
|
||||
|
||||
// NewMCPTokensPurge builds the purger.
|
||||
func NewMCPTokensPurge(repo MCPTokensPurgeRepo, retention time.Duration, log *slog.Logger) *MCPTokensPurge {
|
||||
if log == nil {
|
||||
log = slog.Default()
|
||||
}
|
||||
return &MCPTokensPurge{repo: repo, retention: retention, log: log}
|
||||
}
|
||||
|
||||
// Run purges once.
|
||||
func (r *MCPTokensPurge) Run(ctx context.Context) {
|
||||
cutoff := time.Now().Add(-r.retention)
|
||||
rows, err := r.repo.PurgeExpired(ctx, cutoff)
|
||||
if err != nil {
|
||||
r.log.Error("mcp_tokens.purge", "err", err.Error())
|
||||
return
|
||||
}
|
||||
if rows > 0 {
|
||||
r.log.Info("mcp_tokens.purged", "count", rows, "before", cutoff.UTC().Format(time.RFC3339))
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,62 @@
|
||||
package runners_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/yan1h/agent-coding-workflow/internal/jobs/runners"
|
||||
)
|
||||
|
||||
type fakeMCPTokensPurgeRepo struct {
|
||||
purged int64
|
||||
purgeErr error
|
||||
capturedCutoff time.Time
|
||||
called bool
|
||||
}
|
||||
|
||||
func (f *fakeMCPTokensPurgeRepo) PurgeExpired(_ context.Context, before time.Time) (int64, error) {
|
||||
f.called = true
|
||||
f.capturedCutoff = before
|
||||
if f.purgeErr != nil {
|
||||
return 0, f.purgeErr
|
||||
}
|
||||
return f.purged, nil
|
||||
}
|
||||
|
||||
func TestMCPTokensPurge_RunCallsPurgeWithCorrectCutoff(t *testing.T) {
|
||||
t.Parallel()
|
||||
repo := &fakeMCPTokensPurgeRepo{purged: 4}
|
||||
retention := 168 * time.Hour
|
||||
r := runners.NewMCPTokensPurge(repo, retention, discardLogger())
|
||||
|
||||
before := time.Now()
|
||||
r.Run(context.Background())
|
||||
after := time.Now()
|
||||
|
||||
require.True(t, repo.called)
|
||||
expectedFloor := before.Add(-retention).Add(-time.Second)
|
||||
expectedCeil := after.Add(-retention).Add(time.Second)
|
||||
assert.True(t, !repo.capturedCutoff.Before(expectedFloor) && !repo.capturedCutoff.After(expectedCeil),
|
||||
"cutoff %s outside [%s, %s]", repo.capturedCutoff, expectedFloor, expectedCeil)
|
||||
}
|
||||
|
||||
func TestMCPTokensPurge_RunErrorDoesNotPanic(t *testing.T) {
|
||||
t.Parallel()
|
||||
repo := &fakeMCPTokensPurgeRepo{purgeErr: errors.New("db down")}
|
||||
r := runners.NewMCPTokensPurge(repo, time.Hour, discardLogger())
|
||||
require.NotPanics(t, func() { r.Run(context.Background()) })
|
||||
assert.True(t, repo.called)
|
||||
}
|
||||
|
||||
func TestMCPTokensPurge_NoRowsIsNoOp(t *testing.T) {
|
||||
t.Parallel()
|
||||
repo := &fakeMCPTokensPurgeRepo{purged: 0}
|
||||
r := runners.NewMCPTokensPurge(repo, time.Hour, discardLogger())
|
||||
require.NotPanics(t, func() { r.Run(context.Background()) })
|
||||
assert.True(t, repo.called)
|
||||
}
|
||||
Reference in New Issue
Block a user