You've already forked agentic-coding-workflow
feat(chat): EndpointService for admin LLM endpoints/models with cache invalidation + EndpointLoader
This commit is contained in:
@@ -0,0 +1,266 @@
|
||||
package chat
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
|
||||
"github.com/google/uuid"
|
||||
|
||||
"github.com/yan1h/agent-coding-workflow/internal/audit"
|
||||
"github.com/yan1h/agent-coding-workflow/internal/infra/crypto"
|
||||
"github.com/yan1h/agent-coding-workflow/internal/infra/errs"
|
||||
"github.com/yan1h/agent-coding-workflow/internal/infra/llm"
|
||||
)
|
||||
|
||||
// compile-time assertions
|
||||
var _ EndpointService = (*endpointService)(nil)
|
||||
var _ llm.EndpointLoader = (*endpointLoader)(nil)
|
||||
|
||||
// ===== endpointService =====
|
||||
|
||||
type endpointService struct {
|
||||
repo Repository
|
||||
crypto *crypto.Encryptor
|
||||
registry *llm.Registry
|
||||
auditor audit.Recorder
|
||||
log *slog.Logger
|
||||
}
|
||||
|
||||
// NewEndpointService constructs an EndpointService.
|
||||
func NewEndpointService(repo Repository, c *crypto.Encryptor, reg *llm.Registry, a audit.Recorder, log *slog.Logger) EndpointService {
|
||||
return &endpointService{repo: repo, crypto: c, registry: reg, auditor: a, log: log}
|
||||
}
|
||||
|
||||
// ListEndpoints returns all LLM endpoints.
|
||||
func (s *endpointService) ListEndpoints(ctx context.Context) ([]LLMEndpoint, error) {
|
||||
return s.repo.ListLLMEndpoints(ctx)
|
||||
}
|
||||
|
||||
// CreateEndpoint encrypts the API key, persists the endpoint, and emits an audit event.
|
||||
func (s *endpointService) CreateEndpoint(ctx context.Context, userID uuid.UUID, in CreateEndpointInput) (*LLMEndpoint, error) {
|
||||
enc, err := s.crypto.Encrypt([]byte(in.APIKey))
|
||||
if err != nil {
|
||||
return nil, errs.Wrap(err, errs.CodeInternal, "encrypt api key")
|
||||
}
|
||||
|
||||
id, err := uuid.NewV7()
|
||||
if err != nil {
|
||||
return nil, errs.Wrap(err, errs.CodeInternal, "generate endpoint id")
|
||||
}
|
||||
|
||||
ep, err := s.repo.InsertLLMEndpoint(ctx, InsertEndpointParams{
|
||||
ID: id,
|
||||
Provider: in.Provider,
|
||||
DisplayName: in.DisplayName,
|
||||
BaseURL: in.BaseURL,
|
||||
APIKeyEncrypted: enc,
|
||||
CreatedBy: userID,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
s.tryAudit(ctx, audit.Entry{
|
||||
UserID: &userID,
|
||||
Action: "llm_endpoint.created",
|
||||
TargetType: "llm_endpoint",
|
||||
TargetID: ep.ID.String(),
|
||||
})
|
||||
return ep, nil
|
||||
}
|
||||
|
||||
// UpdateEndpoint updates mutable fields and invalidates the registry cache.
|
||||
func (s *endpointService) UpdateEndpoint(ctx context.Context, id uuid.UUID, in UpdateEndpointInput) error {
|
||||
// If a new API key is provided, encrypt it first; the repo layer converts
|
||||
// in.APIKey (*string) back to []byte, so we pass the encrypted bytes as a
|
||||
// string (binary-safe round-trip in Go).
|
||||
if in.APIKey != nil {
|
||||
enc, err := s.crypto.Encrypt([]byte(*in.APIKey))
|
||||
if err != nil {
|
||||
return errs.Wrap(err, errs.CodeInternal, "encrypt api key")
|
||||
}
|
||||
encStr := string(enc)
|
||||
in.APIKey = &encStr
|
||||
}
|
||||
|
||||
if err := s.repo.UpdateLLMEndpoint(ctx, id, in); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
s.registry.Invalidate(id)
|
||||
|
||||
s.tryAudit(ctx, audit.Entry{
|
||||
Action: "llm_endpoint.updated",
|
||||
TargetType: "llm_endpoint",
|
||||
TargetID: id.String(),
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
// DeleteEndpoint removes an endpoint and invalidates the registry cache.
|
||||
func (s *endpointService) DeleteEndpoint(ctx context.Context, id uuid.UUID) error {
|
||||
if err := s.repo.DeleteLLMEndpoint(ctx, id); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
s.registry.Invalidate(id)
|
||||
|
||||
s.tryAudit(ctx, audit.Entry{
|
||||
Action: "llm_endpoint.deleted",
|
||||
TargetType: "llm_endpoint",
|
||||
TargetID: id.String(),
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
// TestEndpoint pings the upstream by streaming a minimal request.
|
||||
func (s *endpointService) TestEndpoint(ctx context.Context, id uuid.UUID) error {
|
||||
client, err := s.registry.GetClient(ctx, id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Find at least one enabled model for this endpoint.
|
||||
models, err := s.repo.ListLLMModels(ctx, true)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var modelID string
|
||||
for _, m := range models {
|
||||
if m.EndpointID == id {
|
||||
modelID = m.ModelID
|
||||
break
|
||||
}
|
||||
}
|
||||
if modelID == "" {
|
||||
return errs.New(errs.CodeInvalidInput, "no enabled model under this endpoint")
|
||||
}
|
||||
|
||||
ch, err := client.Stream(ctx, modelID, llm.Request{
|
||||
Messages: []llm.Message{
|
||||
{Role: llm.RoleUser, Blocks: []llm.ContentBlock{{Type: "text", Text: "ping"}}},
|
||||
},
|
||||
MaxOutputTokens: 4,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for ev := range ch {
|
||||
switch ev.Type {
|
||||
case llm.EventError:
|
||||
return ev.Err
|
||||
case llm.EventDone:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ListModels returns LLM models, optionally filtered to enabled-only.
|
||||
func (s *endpointService) ListModels(ctx context.Context, onlyEnabled bool) ([]LLMModel, error) {
|
||||
return s.repo.ListLLMModels(ctx, onlyEnabled)
|
||||
}
|
||||
|
||||
// CreateModel persists a new model and emits an audit event.
|
||||
func (s *endpointService) CreateModel(ctx context.Context, in CreateModelInput) (*LLMModel, error) {
|
||||
id, err := uuid.NewV7()
|
||||
if err != nil {
|
||||
return nil, errs.Wrap(err, errs.CodeInternal, "generate model id")
|
||||
}
|
||||
|
||||
m, err := s.repo.InsertLLMModel(ctx, InsertModelParams{
|
||||
ID: id,
|
||||
EndpointID: in.EndpointID,
|
||||
ModelID: in.ModelID,
|
||||
DisplayName: in.DisplayName,
|
||||
Capabilities: in.Capabilities,
|
||||
ContextWindow: in.ContextWindow,
|
||||
MaxOutputTokens: in.MaxOutputTokens,
|
||||
PromptPrice: in.PromptPrice,
|
||||
CompletionPrice: in.CompletionPrice,
|
||||
ThinkingPrice: in.ThinkingPrice,
|
||||
IsTitleGenerator: in.IsTitleGenerator,
|
||||
Enabled: in.Enabled,
|
||||
SortOrder: in.SortOrder,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
s.tryAudit(ctx, audit.Entry{
|
||||
Action: "llm_model.created",
|
||||
TargetType: "llm_model",
|
||||
TargetID: m.ID.String(),
|
||||
})
|
||||
return m, nil
|
||||
}
|
||||
|
||||
// UpdateModel updates mutable model fields.
|
||||
func (s *endpointService) UpdateModel(ctx context.Context, id uuid.UUID, in UpdateModelInput) error {
|
||||
if err := s.repo.UpdateLLMModel(ctx, id, in); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
s.tryAudit(ctx, audit.Entry{
|
||||
Action: "llm_model.updated",
|
||||
TargetType: "llm_model",
|
||||
TargetID: id.String(),
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
// DeleteModel removes a model.
|
||||
func (s *endpointService) DeleteModel(ctx context.Context, id uuid.UUID) error {
|
||||
if err := s.repo.DeleteLLMModel(ctx, id); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
s.tryAudit(ctx, audit.Entry{
|
||||
Action: "llm_model.deleted",
|
||||
TargetType: "llm_model",
|
||||
TargetID: id.String(),
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
// tryAudit logs a warning on failure but never fails the calling operation.
|
||||
func (s *endpointService) tryAudit(ctx context.Context, e audit.Entry) {
|
||||
if err := s.auditor.Record(ctx, e); err != nil {
|
||||
s.log.Warn("audit record failed", "action", e.Action, "err", err)
|
||||
}
|
||||
}
|
||||
|
||||
// ===== endpointLoader =====
|
||||
|
||||
// endpointLoader implements llm.EndpointLoader. Used by app.go to wire the
|
||||
// llm.Registry: it pulls a chat.LLMEndpoint via Repository, decrypts the
|
||||
// API key, and returns an llm.Endpoint suitable for buildClient.
|
||||
type endpointLoader struct {
|
||||
repo Repository
|
||||
crypto *crypto.Encryptor
|
||||
}
|
||||
|
||||
// NewEndpointLoader constructs an llm.EndpointLoader bound to repo + crypto.
|
||||
func NewEndpointLoader(repo Repository, c *crypto.Encryptor) llm.EndpointLoader {
|
||||
return &endpointLoader{repo: repo, crypto: c}
|
||||
}
|
||||
|
||||
func (l *endpointLoader) Load(ctx context.Context, id uuid.UUID) (llm.Endpoint, error) {
|
||||
ep, err := l.repo.GetLLMEndpoint(ctx, id)
|
||||
if err != nil {
|
||||
return llm.Endpoint{}, err
|
||||
}
|
||||
plain, err := l.crypto.Decrypt(ep.APIKeyEncrypted)
|
||||
if err != nil {
|
||||
return llm.Endpoint{}, fmt.Errorf("decrypt api key: %w", err)
|
||||
}
|
||||
return llm.Endpoint{
|
||||
ID: ep.ID,
|
||||
Provider: string(ep.Provider),
|
||||
BaseURL: ep.BaseURL,
|
||||
APIKey: string(plain),
|
||||
}, nil
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user