refreshpg

README

auth/refreshpg

Postgres-backed auth.RefreshStore поверх db.Querier. Атомарный Consume через единый UPDATE … RETURNING; reuse detection триггерит family-wide RevokeFamily перед возвратом *errs.Error{Code: "refresh_reused"}. DDL живёт в schema.sql — сам пакет миграции не выполняет.

Родитель: ../README.md Импорт: github.com/theizzatbek/gokit/auth/refreshpg

Использование

import (
    "github.com/theizzatbek/gokit/auth"
    "github.com/theizzatbek/gokit/auth/refreshpg"
    "github.com/theizzatbek/gokit/db"
)

d, _ := db.Connect(ctx, dbCfg)

authObj, _ := auth.New[MyClaims](auth.Config{
    Issuer: "myservice", Keys: ks, AccessTTL: 15*time.Minute, RefreshTTL: 30*24*time.Hour,
}, auth.WithRefreshStore(refreshpg.New(d)))

refreshpg.New(d) принимает любой db.Querier — *db.DB или *db.Tx. Тесты могут передать транзакцию, чтобы изменения откатывались на конце теста.

Схема

Примените schema.sql (или скопируйте DDL в свою миграцию) перед первым использованием:

CREATE TABLE IF NOT EXISTS auth_refresh_tokens (
    token_hash  bytea       PRIMARY KEY,
    family_id   uuid        NOT NULL,
    parent_hash bytea       NOT NULL,
    subject     text        NOT NULL,
    issued_at   timestamptz NOT NULL,
    expires_at  timestamptz NOT NULL,
    consumed_at timestamptz,
    revoked_at  timestamptz,
    user_agent  text        NOT NULL DEFAULT '',
    ip          inet
);
CREATE INDEX IF NOT EXISTS auth_refresh_tokens_family_id_idx ON auth_refresh_tokens (family_id);
CREATE INDEX IF NOT EXISTS auth_refresh_tokens_subject_idx   ON auth_refresh_tokens (subject);
CREATE INDEX IF NOT EXISTS auth_refresh_tokens_expires_at_idx ON auth_refresh_tokens (expires_at);

examples/urlshort/migrations/0001_init.sql включает этот DDL дословно рядом с собственными таблицами сервиса.

Admin / operator API

*Store несёт ряд методов вне auth.RefreshStore-интерфейса — для admin-эндпоинтов, инцидент-респонса и cron-cleanup'а. Доступны на типе *refreshpg.Store напрямую.

Метод	Возвращает	Заметки
ListBySubject(ctx, subject) ([]SessionInfo, error)	history sessions	Все строки subject ordered issued_at DESC; включает active / consumed / revoked / expired (UI фильтрует по State).
Stats(ctx) (StoreStats, error)	{Active, Consumed, Revoked, Expired, Total}	Disjoint buckets, один round trip.
RevokeByIP(ctx, ip) (int64, error)	число revoked tokens	Bulk-revoke по IP-адресу для incident response. Empty ip → 0; unknown ip → 0 (idempotent).
GarbageCollectBatch(ctx, now, limit, maxIterations)	число удалённых	Chunked variant of GarbageCollect для очень больших таблиц; loop DELETE … LIMIT N пока не вернёт 0. limit ≤ 0 → 1000, maxIterations ≤ 0 → 1024. На ctx cancel возвращает частичный прогресс.

SessionInfo ничего не несёт из token_hash — секрет никогда не покидает store. Поле State — "active" | "consumed" | "revoked" | "expired" (disjoint).

Observability + хуки

refreshpg.New(d, opts...) принимает функциональные опции (обратно совместимо с refreshpg.New(d)):

• WithMetrics(reg prometheus.Registerer) регистрирует:
  • refreshpg_ops_total{op,outcome} — Issue / Consume / RevokeFamily / RevokeSubject / RevokeByIP / GC / Stats / List; outcome — ok | error (consume также: missing | expired | reused).
  • refreshpg_op_duration_seconds{op} — histogram wall-clock latency.
• WithLogger(*slog.Logger) — silent по умолчанию; используется только для panic-recovery в hooks и диагностических warning'ов.
• WithOnConsumeReused(fn) — fires ВНУТРИ Consume после reuse-detection (RevokeFamily уже отработал). Подключите к SIEM / Sentry — это OAuth 2.1 stolen-token alert.
• WithOnFamilyRevoke(fn) / WithOnSubjectRevoke(fn) / WithOnIPRevoke(fn) — post-revoke audit hooks. Все panic-safe (recovered + WARN-logged через WithLogger).

Заметки

• Хеши токенов, а не сами токены. Сырой refresh token никогда не попадает в БД — только sha256(token). Утечка БД не компрометирует активные refresh-токены.
• Family revoke при reuse. Когда Consume видит токен, у которого consumed_at IS NOT NULL, он делает RevokeFamily(family_id) перед возвратом ошибки. Это каноническая реакция на "stolen-token detected": invalidate каждого потомка скомпрометированного root-токена.
• Никакой фоновой чистки expired. Истёкшие строки остаются в таблице. Запускайте GarbageCollect(ctx, now) (одним DELETE) или GarbageCollectBatch(ctx, now, 1000, 0) (chunked) из nightly cron'а, если хотите освобождать место.
• Атомарно через UPDATE … RETURNING. Никакого race window SELECT-then-UPDATE. Диагностический SELECT на miss-пути классифицирует, существовал ли токен вообще или уже был consumed.
• SecurityLogger на *auth.Auth (через auth.WithSecurityLogger) эмитит структурированные WARN-события для reuse-triggered revocations — подключите к вашему SIEM/alerting.

Тестирование

Используйте testcontainers-go/modules/postgres. Паттерн из store_test.go самого gokit:

ctx := context.Background()
c, _ := tcpostgres.Run(ctx, "postgres:16-alpine",
    tcpostgres.WithDatabase("test"), tcpostgres.WithUsername("test"), tcpostgres.WithPassword("test"),
    tcpostgres.BasicWaitStrategies())
defer testcontainers.TerminateContainer(c)

d, _ := db.Connect(ctx, /* derive cfg from c */)
_, _ = d.Exec(ctx, refreshpg.SchemaSQL())  // или inline DDL
store := refreshpg.New(d)

См. также

• auth — родитель: WithRefreshStore потребляет это
• auth/refreshredis — тот же контракт, Redis-backed
• db — предоставляет интерфейс Querier

Documentation

Overview

Package refreshpg implements auth.RefreshStore on top of Postgres via the kit's db.Querier interface. The DDL lives in schema.sql next to this file; downstream services run it through their own migration tool.

Index

• type ConsumeReusedHook
• type FamilyRevokeHook
• type IPRevokeHook
• type Option
  • func WithLogger(l *slog.Logger) Option
  • func WithMetrics(reg prometheus.Registerer) Option
  • func WithOnConsumeReused(fn ConsumeReusedHook) Option
  • func WithOnFamilyRevoke(fn FamilyRevokeHook) Option
  • func WithOnIPRevoke(fn IPRevokeHook) Option
  • func WithOnSubjectRevoke(fn SubjectRevokeHook) Option
• type SessionInfo
• type Store
  • func New(q db.Querier, opts ...Option) *Store
  • func (s *Store) Consume(ctx context.Context, tokenHash [32]byte, now time.Time) (auth.Record, error)
  • func (s *Store) GarbageCollect(ctx context.Context, now time.Time) (int64, error)
  • func (s *Store) GarbageCollectBatch(ctx context.Context, now time.Time, limit, maxIterations int) (int64, error)
  • func (s *Store) Issue(ctx context.Context, r auth.Record) error
  • func (s *Store) ListBySubject(ctx context.Context, subject string) ([]SessionInfo, error)
  • func (s *Store) RevokeByIP(ctx context.Context, ip string) (int64, error)
  • func (s *Store) RevokeFamily(ctx context.Context, familyID string) error
  • func (s *Store) RevokeSubject(ctx context.Context, subject string) error
  • func (s *Store) RevokeToken(ctx context.Context, tokenHash [32]byte, now time.Time) (auth.Record, bool, error)
  • func (s *Store) Stats(ctx context.Context) (StoreStats, error)
• type StoreStats
• type SubjectRevokeHook

Types

type ConsumeReusedHook

type ConsumeReusedHook func(ctx context.Context, familyID, subject string)

ConsumeReusedHook fires INSIDE Consume when the kit detects a refresh token that has already been consumed or revoked — the OAuth 2.1 reuse-detection signal. RevokeFamily runs FIRST (the security-critical side effect happens regardless of hook behaviour); the hook only sees the event after the family has already been wiped.

Wire it to your SIEM / Sentry / pager — a non-zero rate is the canonical "stolen refresh token" alert. Panic-safe.

type FamilyRevokeHook

type FamilyRevokeHook func(ctx context.Context, familyID string, count int64)

FamilyRevokeHook fires after a successful RevokeFamily. `count` is the number of rows revoked (0 = idempotent no-op). Panic-safe.

type IPRevokeHook

type IPRevokeHook func(ctx context.Context, ip string, count int64)

IPRevokeHook fires after a successful RevokeByIP. Panic-safe. Use for incident-response audit trails.

type Option

type Option func(*storeOpts)

Option tunes the constructed *Store. The variadic shape on New keeps the zero-option call site (`refreshpg.New(db)`) compatible.

func WithLogger

func WithLogger(l *slog.Logger) Option

WithLogger wires a slog logger. Used for hook panic recovery and (sparingly) for diagnostic warnings — store ops are otherwise silent. nil = silent.

func WithMetrics

func WithMetrics(reg prometheus.Registerer) Option

WithMetrics enables Prometheus instrumentation. The store registers:

• refreshpg_ops_total{op,outcome} — Issue / Consume / RevokeFamily / RevokeSubject / RevokeByIP / GarbageCollect / Stats / ListBySubject; outcome=ok|error (consume also: missing| expired|reused)
• refreshpg_op_duration_seconds{op} — wall-clock latency

Pass the same Registerer you give to other kit subsystems for a single /metrics scrape.

func WithOnConsumeReused

func WithOnConsumeReused(fn ConsumeReusedHook) Option

WithOnConsumeReused registers a callback fired on every refresh reuse-detection event. See ConsumeReusedHook. Multiple calls — last wins.

func WithOnFamilyRevoke

func WithOnFamilyRevoke(fn FamilyRevokeHook) Option

WithOnFamilyRevoke registers a post-RevokeFamily callback. See FamilyRevokeHook. Multiple calls — last wins.

func WithOnIPRevoke

func WithOnIPRevoke(fn IPRevokeHook) Option

WithOnIPRevoke registers a post-RevokeByIP callback. See IPRevokeHook. Multiple calls — last wins.

func WithOnSubjectRevoke

func WithOnSubjectRevoke(fn SubjectRevokeHook) Option

WithOnSubjectRevoke registers a post-RevokeSubject callback. See SubjectRevokeHook. Multiple calls — last wins.

type SessionInfo

type SessionInfo struct {
	FamilyID   string
	Subject    string
	IssuedAt   time.Time
	ExpiresAt  time.Time
	ConsumedAt time.Time // zero = still live
	RevokedAt  time.Time // zero = active
	UserAgent  string
	IP         string
	State      string // "active" | "consumed" | "revoked" | "expired"
}

SessionInfo is the admin-side projection of an issued refresh-token row. Returned by Store.ListBySubject. NEVER includes `token_hash` — the hash is secret material and stays in the store.

type Store

type Store struct {
	// contains filtered or unexported fields
}

Store is the Postgres-backed RefreshStore. Pool ownership stays with the caller. Observability + lifecycle hooks are opt-in via Option.

func New

func New(q db.Querier, opts ...Option) *Store

New takes any db.Querier (so callers can pass a *db.DB or a *db.Tx). The most common case — long-lived service — passes *db.DB. Trailing options enable metrics / logging / hooks; the zero-option form is unchanged from earlier versions.

func (*Store) Consume

func (s *Store) Consume(ctx context.Context, tokenHash [32]byte, now time.Time) (auth.Record, error)

Consume atomically marks a refresh token consumed if it is live, valid, and not expired; otherwise it diagnoses the failure mode and (for reuse) revokes the whole family before returning the error.

The implementation is two queries: a single UPDATE ... RETURNING that succeeds for the live-and-valid case, and a follow-up SELECT only in the failure case to disambiguate not_found / expired / reused.

func (*Store) GarbageCollect

func (s *Store) GarbageCollect(ctx context.Context, now time.Time) (int64, error)

GarbageCollect deletes records with expires_at <= now. Returns the number of rows removed.

func (*Store) GarbageCollectBatch

func (s *Store) GarbageCollectBatch(ctx context.Context, now time.Time, limit, maxIterations int) (int64, error)

GarbageCollectBatch is the chunked variant of Store.GarbageCollect for very large tables: it loops `DELETE … WHERE expires_at <= now LIMIT N` until the affected count is zero, returning the total number of rows removed. Use when Store.GarbageCollect would lock the table for too long under nightly cron pressure.

`limit ≤ 0` is treated as 1000. `maxIterations ≤ 0` defaults to 1024 (a defence against pathological GC loops). On ctx cancel the method returns whatever it has already deleted plus ctx.Err() — partial progress is the safe behaviour for a sweeper.

func (*Store) Issue

func (s *Store) Issue(ctx context.Context, r auth.Record) error

Issue inserts a row. pgx errors are funneled through db/'s mapPgxErr (via the Querier), so a unique-key collision returns *errs.Error{KindAlreadyExists}.

func (*Store) ListBySubject

func (s *Store) ListBySubject(ctx context.Context, subject string) ([]SessionInfo, error)

ListBySubject returns every refresh-token row for the subject ordered by `issued_at DESC`. Surface this to admin UIs that render an "active sessions" list — UI filters by `State` field to hide history.

Empty subject returns an empty slice without touching the DB.

func (*Store) RevokeByIP

func (s *Store) RevokeByIP(ctx context.Context, ip string) (int64, error)

RevokeByIP revokes every currently-active refresh token issued from the supplied IP address. Use for incident response — a leaked session cookie or a compromised egress IP — to evict every active family bound to that address in one round trip.

Returns the number of rows revoked. Zero is NOT an error (idempotent against repeat calls and against IPs that never issued). Empty `ip` returns 0 without touching the DB.

The supplied string is parsed as an `inet` literal by Postgres — callers can pass either a v4 address ("203.0.113.7") or a v6 ("2001:db8::1"). Mask/CIDR forms are not supported (single-host semantics are what incident response actually wants).

func (*Store) RevokeFamily

func (s *Store) RevokeFamily(ctx context.Context, familyID string) error

RevokeFamily marks every live token in the family as revoked. Idempotent via `COALESCE(revoked_at, now())` + `WHERE revoked_at IS NULL`.

func (*Store) RevokeSubject

func (s *Store) RevokeSubject(ctx context.Context, subject string) error

RevokeSubject revokes every live token belonging to the subject. Idempotent.

func (*Store) RevokeToken

func (s *Store) RevokeToken(ctx context.Context, tokenHash [32]byte, now time.Time) (auth.Record, bool, error)

RevokeToken marks the single matching record revoked — the store half of auth.TokenRevoker. Idempotent: an already-revoked record keeps its original revoked_at (COALESCE) and is still returned; a missing record is (Record{}, false, nil). Single UPDATE … RETURNING round trip.

func (*Store) Stats

func (s *Store) Stats(ctx context.Context) (StoreStats, error)

Stats returns the disjoint-bucket rollup in a single round trip. Suitable for /admin or /metrics-pull endpoints; not on a hot path.

type StoreStats

type StoreStats struct {
	Active   int
	Consumed int
	Revoked  int
	Expired  int
	Total    int
}

StoreStats is the rollup returned by Store.Stats. Buckets are disjoint:

Active   = consumed_at IS NULL AND revoked_at IS NULL AND expires_at > NOW()
Consumed = consumed_at IS NOT NULL AND revoked_at IS NULL
Revoked  = revoked_at IS NOT NULL (revoke wins over consume/expiry)
Expired  = consumed_at IS NULL AND revoked_at IS NULL AND expires_at <= NOW()

Total = Active + Consumed + Revoked + Expired.

type SubjectRevokeHook

type SubjectRevokeHook func(ctx context.Context, subject string, count int64)

SubjectRevokeHook fires after a successful RevokeSubject. Panic-safe.