merge: demo wiring — tapir auth/run + config (Worker F, agent/demo-wiring)
# Conflicts: # cmd/tapir/main.go
This commit is contained in:
@@ -17,11 +17,20 @@ import (
|
||||
"time"
|
||||
)
|
||||
|
||||
// defaultMaxTokens is sent on every request. Tapir CHANGES this from the
|
||||
// hyperguild copy (ADR-004 says change the copy, not the upstream): thinking
|
||||
// models (qwen3, deepseek-r1) spend their budget on reasoning and return EMPTY
|
||||
// content when max_tokens is unset or too low. A generous ceiling leaves room
|
||||
// for both the reasoning trace and the actual summary. See
|
||||
// docs/homelab-integration.md.
|
||||
const defaultMaxTokens = 8192
|
||||
|
||||
// Client calls an OpenAI-compatible chat completions endpoint.
|
||||
type Client struct {
|
||||
baseURL string
|
||||
apiKey string
|
||||
model string
|
||||
maxTokens int
|
||||
httpClient *http.Client
|
||||
}
|
||||
|
||||
@@ -31,6 +40,7 @@ func New(baseURL, apiKey, model string, timeout time.Duration) *Client {
|
||||
baseURL: strings.TrimRight(baseURL, "/"),
|
||||
apiKey: apiKey,
|
||||
model: model,
|
||||
maxTokens: defaultMaxTokens,
|
||||
httpClient: &http.Client{Timeout: timeout},
|
||||
}
|
||||
}
|
||||
@@ -39,6 +49,7 @@ type chatRequest struct {
|
||||
Model string `json:"model"`
|
||||
Messages []message `json:"messages"`
|
||||
Temperature float64 `json:"temperature"`
|
||||
MaxTokens int `json:"max_tokens,omitempty"`
|
||||
}
|
||||
|
||||
type message struct {
|
||||
@@ -62,6 +73,7 @@ func (c *Client) Complete(ctx context.Context, system, user string) (string, err
|
||||
{Role: "user", Content: user},
|
||||
},
|
||||
Temperature: 0.2,
|
||||
MaxTokens: c.maxTokens,
|
||||
}
|
||||
b, err := json.Marshal(body)
|
||||
if err != nil {
|
||||
|
||||
@@ -43,6 +43,27 @@ func TestClient_Complete(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestClient_SendsMaxTokens guards Tapir's ADR-004 change to the copied client:
|
||||
// it MUST send a positive max_tokens, or thinking models return empty content.
|
||||
func TestClient_SendsMaxTokens(t *testing.T) {
|
||||
var body chatRequest
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
_ = json.NewDecoder(r.Body).Decode(&body)
|
||||
_ = json.NewEncoder(w).Encode(map[string]any{
|
||||
"choices": []map[string]any{{"message": map[string]any{"content": "ok"}}},
|
||||
})
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
c := New(srv.URL, "", "test-model", 10*time.Second)
|
||||
if _, err := c.Complete(context.Background(), "sys", "user"); err != nil {
|
||||
t.Fatalf("Complete: %v", err)
|
||||
}
|
||||
if body.MaxTokens <= 0 {
|
||||
t.Errorf("max_tokens = %d, want > 0 (thinking models return empty content without it)", body.MaxTokens)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClient_ReturnsErrorOnNon200(t *testing.T) {
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
http.Error(w, "overloaded", http.StatusServiceUnavailable)
|
||||
|
||||
@@ -0,0 +1,107 @@
|
||||
// Package secrets provides a local file-backed implementation of the
|
||||
// ports.SecretStore port. It is a Stage-0 stand-in for op/ESO: secret material
|
||||
// (the YouTube OAuth refresh token) is kept in a 0600 JSON file rather than the
|
||||
// vault, so the demo runs without live op. Because every consumer depends on
|
||||
// the SecretStore port, swapping this for an op/ESO-backed store later is a
|
||||
// wiring change, not a code change (ADR-002, docs/homelab-integration.md).
|
||||
//
|
||||
// Secret values are never logged. Get returns an error for an unknown ref so a
|
||||
// missing token surfaces loudly rather than as an empty string.
|
||||
package secrets
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
|
||||
"gitea.d-ma.be/mathias/tapir/internal/ports"
|
||||
)
|
||||
|
||||
// FileStore is a SecretStore backed by a single 0600 JSON file mapping opaque
|
||||
// refs to secret values. Safe for concurrent use within one process.
|
||||
type FileStore struct {
|
||||
path string
|
||||
mu sync.RWMutex
|
||||
}
|
||||
|
||||
// Static check: FileStore satisfies the read side of the port.
|
||||
var _ ports.SecretStore = (*FileStore)(nil)
|
||||
|
||||
// ErrNotFound is returned by Get when no secret is stored under the ref.
|
||||
var ErrNotFound = errors.New("secrets: ref not found")
|
||||
|
||||
// NewFileStore returns a store backed by path. The file need not exist yet; it
|
||||
// is created on the first Put.
|
||||
func NewFileStore(path string) *FileStore {
|
||||
return &FileStore{path: path}
|
||||
}
|
||||
|
||||
// Get resolves a ref to its secret value. It returns ErrNotFound if the file or
|
||||
// the ref is absent.
|
||||
func (s *FileStore) Get(_ context.Context, ref string) (string, error) {
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
|
||||
m, err := s.load()
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
v, ok := m[ref]
|
||||
if !ok {
|
||||
return "", fmt.Errorf("%w: %q", ErrNotFound, ref)
|
||||
}
|
||||
return v, nil
|
||||
}
|
||||
|
||||
// Put stores value under ref, persisting the file with 0600 permissions. It
|
||||
// merges into any existing entries and writes atomically (temp file + rename).
|
||||
func (s *FileStore) Put(ref, value string) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
m, err := s.load()
|
||||
if err != nil && !errors.Is(err, os.ErrNotExist) {
|
||||
return err
|
||||
}
|
||||
if m == nil {
|
||||
m = make(map[string]string)
|
||||
}
|
||||
m[ref] = value
|
||||
|
||||
if err := os.MkdirAll(filepath.Dir(s.path), 0o700); err != nil {
|
||||
return fmt.Errorf("secrets: create dir: %w", err)
|
||||
}
|
||||
b, err := json.Marshal(m)
|
||||
if err != nil {
|
||||
return fmt.Errorf("secrets: marshal: %w", err)
|
||||
}
|
||||
tmp := s.path + ".tmp"
|
||||
if err := os.WriteFile(tmp, b, 0o600); err != nil {
|
||||
return fmt.Errorf("secrets: write temp: %w", err)
|
||||
}
|
||||
if err := os.Rename(tmp, s.path); err != nil {
|
||||
return fmt.Errorf("secrets: rename: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// load reads the backing file. A missing file yields an empty map (not an
|
||||
// error) for Get's caller, except Put distinguishes os.ErrNotExist.
|
||||
func (s *FileStore) load() (map[string]string, error) {
|
||||
b, err := os.ReadFile(s.path)
|
||||
if err != nil {
|
||||
if errors.Is(err, os.ErrNotExist) {
|
||||
return map[string]string{}, nil
|
||||
}
|
||||
return nil, fmt.Errorf("secrets: read %s: %w", s.path, err)
|
||||
}
|
||||
var m map[string]string
|
||||
if err := json.Unmarshal(b, &m); err != nil {
|
||||
return nil, fmt.Errorf("secrets: parse %s: %w", s.path, err)
|
||||
}
|
||||
return m, nil
|
||||
}
|
||||
@@ -0,0 +1,72 @@
|
||||
package secrets_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"gitea.d-ma.be/mathias/tapir/internal/adapters/secrets"
|
||||
)
|
||||
|
||||
func TestPutThenGet(t *testing.T) {
|
||||
path := filepath.Join(t.TempDir(), "nested", "secrets.json")
|
||||
s := secrets.NewFileStore(path)
|
||||
|
||||
if err := s.Put("youtube/refresh_token", "rt-123"); err != nil {
|
||||
t.Fatalf("Put: %v", err)
|
||||
}
|
||||
got, err := s.Get(context.Background(), "youtube/refresh_token")
|
||||
if err != nil {
|
||||
t.Fatalf("Get: %v", err)
|
||||
}
|
||||
if got != "rt-123" {
|
||||
t.Errorf("Get = %q, want %q", got, "rt-123")
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetUnknownRef(t *testing.T) {
|
||||
s := secrets.NewFileStore(filepath.Join(t.TempDir(), "secrets.json"))
|
||||
_, err := s.Get(context.Background(), "missing")
|
||||
if !errors.Is(err, secrets.ErrNotFound) {
|
||||
t.Errorf("err = %v, want ErrNotFound", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPutIsOwnerOnly(t *testing.T) {
|
||||
path := filepath.Join(t.TempDir(), "secrets.json")
|
||||
s := secrets.NewFileStore(path)
|
||||
if err := s.Put("k", "v"); err != nil {
|
||||
t.Fatalf("Put: %v", err)
|
||||
}
|
||||
info, err := os.Stat(path)
|
||||
if err != nil {
|
||||
t.Fatalf("Stat: %v", err)
|
||||
}
|
||||
if perm := info.Mode().Perm(); perm != 0o600 {
|
||||
t.Errorf("file perm = %o, want 0600 (token must not be world-readable)", perm)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPutMergesEntries(t *testing.T) {
|
||||
path := filepath.Join(t.TempDir(), "secrets.json")
|
||||
s := secrets.NewFileStore(path)
|
||||
if err := s.Put("a", "1"); err != nil {
|
||||
t.Fatalf("Put a: %v", err)
|
||||
}
|
||||
if err := s.Put("b", "2"); err != nil {
|
||||
t.Fatalf("Put b: %v", err)
|
||||
}
|
||||
// Re-open from disk to prove persistence, not in-memory state.
|
||||
s2 := secrets.NewFileStore(path)
|
||||
for k, want := range map[string]string{"a": "1", "b": "2"} {
|
||||
got, err := s2.Get(context.Background(), k)
|
||||
if err != nil {
|
||||
t.Fatalf("Get %q: %v", k, err)
|
||||
}
|
||||
if got != want {
|
||||
t.Errorf("Get %q = %q, want %q", k, got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,77 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"gitea.d-ma.be/mathias/tapir/internal/domain"
|
||||
)
|
||||
|
||||
// UpsertVideo persists a video's metadata and returns its durable store id (the
|
||||
// videos.id UUID). It is idempotent on (user_id, provider, provider_video_id):
|
||||
// the same provider video for a user always resolves to the same row and the
|
||||
// same returned id, so the run loop can use that id as the stable dedup key
|
||||
// across restarts (it matches summaries.video_id once a summary exists).
|
||||
//
|
||||
// This lives in a separate file from store.go on purpose: the Sink port only
|
||||
// carries a domain.Summary (no title/channel), so video metadata is persisted
|
||||
// here, out of the delivery path, to keep the reader's rows readable.
|
||||
//
|
||||
// subscription_id is intentionally left NULL at Stage 0: the YouTube
|
||||
// Subscription.ID is a provider resource id, not the UUID that column expects,
|
||||
// and the subscriptions table is not part of this slice (data-model.md).
|
||||
func (s *Store) UpsertVideo(ctx context.Context, v domain.Video) (string, error) {
|
||||
if v.UserID == "" {
|
||||
return "", fmt.Errorf("store: upsert video: empty user id")
|
||||
}
|
||||
if v.ProviderVideoID == "" {
|
||||
return "", fmt.Errorf("store: upsert video: empty provider video id")
|
||||
}
|
||||
|
||||
tx, err := s.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("store: begin: %w", err)
|
||||
}
|
||||
defer tx.Rollback(ctx) //nolint:errcheck // no-op after Commit
|
||||
|
||||
// Ensure the owning user exists (FK target) — same as the Deliver path.
|
||||
if _, err := tx.Exec(ctx,
|
||||
`INSERT INTO users (id) VALUES ($1) ON CONFLICT (id) DO NOTHING`,
|
||||
v.UserID); err != nil {
|
||||
return "", fmt.Errorf("store: upsert user: %w", err)
|
||||
}
|
||||
|
||||
provider := string(v.Provider)
|
||||
if provider == "" {
|
||||
provider = string(domain.ProviderYouTube)
|
||||
}
|
||||
|
||||
var id string
|
||||
if err := tx.QueryRow(ctx,
|
||||
`INSERT INTO videos (user_id, provider, provider_video_id, title, url, published_at)
|
||||
VALUES ($1, $2, $3, $4, $5, $6)
|
||||
ON CONFLICT (user_id, provider, provider_video_id) DO UPDATE SET
|
||||
title = EXCLUDED.title,
|
||||
url = EXCLUDED.url,
|
||||
published_at = EXCLUDED.published_at
|
||||
RETURNING id`,
|
||||
v.UserID, provider, v.ProviderVideoID, v.Title, v.URL, nullTime(v.PublishedAt),
|
||||
).Scan(&id); err != nil {
|
||||
return "", fmt.Errorf("store: upsert video: %w", err)
|
||||
}
|
||||
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return "", fmt.Errorf("store: commit: %w", err)
|
||||
}
|
||||
return id, nil
|
||||
}
|
||||
|
||||
// nullTime maps the zero time to NULL so an unknown published_at is stored as
|
||||
// SQL NULL rather than year 0001.
|
||||
func nullTime(t time.Time) *time.Time {
|
||||
if t.IsZero() {
|
||||
return nil
|
||||
}
|
||||
return &t
|
||||
}
|
||||
@@ -0,0 +1,83 @@
|
||||
package store_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"gitea.d-ma.be/mathias/tapir/internal/domain"
|
||||
)
|
||||
|
||||
func ytVideo(userID, provVideoID, title string) domain.Video {
|
||||
return domain.Video{
|
||||
UserID: userID,
|
||||
Provider: domain.ProviderYouTube,
|
||||
ProviderVideoID: provVideoID,
|
||||
Title: title,
|
||||
URL: "https://www.youtube.com/watch?v=" + provVideoID,
|
||||
PublishedAt: time.Date(2026, 6, 1, 12, 0, 0, 0, time.UTC),
|
||||
// SubscriptionID is a provider resource id, not a UUID — must not be
|
||||
// written to the UUID column. Set it to prove UpsertVideo ignores it.
|
||||
SubscriptionID: "yt-subscription-resource-id",
|
||||
}
|
||||
}
|
||||
|
||||
func TestUpsertVideo_ReturnsStableID(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
s := newStore(t)
|
||||
resetDB(t, rawPool(t))
|
||||
|
||||
id1, err := s.UpsertVideo(ctx, ytVideo(userA, "dQw4w9WgXcQ", "first title"))
|
||||
require.NoError(t, err)
|
||||
require.NotEmpty(t, id1)
|
||||
|
||||
// Same (user, provider, provider_video_id) -> same row, same id, updated meta.
|
||||
id2, err := s.UpsertVideo(ctx, ytVideo(userA, "dQw4w9WgXcQ", "updated title"))
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, id1, id2, "idempotent upsert must return the same durable id")
|
||||
|
||||
p := rawPool(t)
|
||||
var (
|
||||
title string
|
||||
count int
|
||||
)
|
||||
require.NoError(t, p.QueryRow(ctx,
|
||||
`SELECT title FROM videos WHERE id = $1`, id1).Scan(&title))
|
||||
require.Equal(t, "updated title", title, "second upsert must update metadata in place")
|
||||
|
||||
require.NoError(t, p.QueryRow(ctx,
|
||||
`SELECT count(*) FROM videos WHERE user_id = $1`, userA).Scan(&count))
|
||||
require.Equal(t, 1, count, "must not duplicate the row")
|
||||
}
|
||||
|
||||
func TestUpsertVideo_IDMatchesSummaryDedup(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
s := newStore(t)
|
||||
resetDB(t, rawPool(t))
|
||||
|
||||
// Upsert assigns the durable video id; a summary delivered under that id
|
||||
// must then show up in SeenVideoIDs — this is the cross-restart dedup chain.
|
||||
id, err := s.UpsertVideo(ctx, ytVideo(userA, "abc123", "t"))
|
||||
require.NoError(t, err)
|
||||
|
||||
require.NoError(t, s.Deliver(ctx, summary(userA, id, "the summary")))
|
||||
|
||||
seen, err := s.SeenVideoIDs(ctx, userA)
|
||||
require.NoError(t, err)
|
||||
require.True(t, seen[id], "the upserted video id must match the summary dedup key")
|
||||
}
|
||||
|
||||
func TestUpsertVideo_PerUserIsolation(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
s := newStore(t)
|
||||
resetDB(t, rawPool(t))
|
||||
|
||||
idA, err := s.UpsertVideo(ctx, ytVideo(userA, "same-provider-id", "a"))
|
||||
require.NoError(t, err)
|
||||
idB, err := s.UpsertVideo(ctx, ytVideo(userB, "same-provider-id", "b"))
|
||||
require.NoError(t, err)
|
||||
|
||||
require.NotEqual(t, idA, idB, "same provider video for two users must be two distinct rows")
|
||||
}
|
||||
Reference in New Issue
Block a user