sendnrw/pocket-id
2
0
Fork 0
mirror of https://github.com/pocket-id/pocket-id.git synced 2026-09-23 19:39:04 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' into feat/huma-api-docs

Browse Source
This commit is contained in:
Kyle Mendell
2026-07-12 12:48:50 -05:00
committed by GitHub
parent f8b441fb06 7348bdac50
commit 3893376cba
44 changed files with 705 additions and 274 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/internal/bootstrap/actors_bootstrap.go
+40 -31
View File
@@ -18,7 +18,6 @@ import (
"github.com/pocket-id/pocket-id/backend/internal/common"
"github.com/pocket-id/pocket-id/backend/internal/job"
"github.com/pocket-id/pocket-id/backend/internal/middleware"
"github.com/pocket-id/pocket-id/backend/internal/service"
"github.com/pocket-id/pocket-id/backend/internal/storage"
"github.com/pocket-id/pocket-id/backend/internal/utils/crypto"
)
@@ -28,7 +27,7 @@ type NewActorsOpts struct {
Postgres *pgxpool.Pool
EnvConfig *common.EnvConfigSchema
AppConfig *service.AppConfigService
InstanceID string
HttpClient *http.Client
DB *gorm.DB
FileStorage storage.FileStorage
@@ -52,20 +51,6 @@ func NewActors(o NewActorsOpts) (*local.Host, map[string]*ratelimit.RateLimitSer
local.WithShutdownGracePeriod(10 * time.Second),
}
// Add all cron jobs
cronjobs, err := o.getCronJobs()
if err != nil {
return nil, nil, err
}
opts = append(opts, cronjobs...)
// Add the rate limiters
rateLimiters, rateLimiterOpts, err := o.getRateLimiters()
if err != nil {
return nil, nil, err
}
opts = append(opts, rateLimiterOpts...)
// Add the database connection
providerOpt, err := o.getProvider()
if err != nil {
@@ -79,6 +64,18 @@ func NewActors(o NewActorsOpts) (*local.Host, map[string]*ratelimit.RateLimitSer
return nil, nil, fmt.Errorf("failed to create actor host: %w", err)
}
// Add all cron jobs
err = o.registerCronJobs(h)
if err != nil {
return nil, nil, err
}
// Add the rate limiters
rateLimiters, err := o.registerRateLimiters(h)
if err != nil {
return nil, nil, err
}
// Bind a service for each rate limiter so the middleware can invoke them
rateLimitServices := make(map[string]*ratelimit.RateLimitService, len(rateLimiters))
for name, rl := range rateLimiters {
@@ -90,8 +87,9 @@ func NewActors(o NewActorsOpts) (*local.Host, map[string]*ratelimit.RateLimitSer
// Derive a PSK from the global encryption key
func (o *NewActorsOpts) getPSK() ([]byte, error) {
// This is tied to the instance ID of the Pocket ID deployment/cluster
// Note: changing the key derivation or the seed is a breaking change
return crypto.DeriveKey(o.EnvConfig.EncryptionKey, "pocketid/actors-psk")
return crypto.DeriveKey(o.EnvConfig.EncryptionKey, "pocketid/actors-psk/"+o.InstanceID)
}
func (o *NewActorsOpts) getProvider() (local.HostOption, error) {
@@ -111,37 +109,44 @@ func (o *NewActorsOpts) getProvider() (local.HostOption, error) {
}
}
func (o *NewActorsOpts) getCronJobs() (opts []local.HostOption, err error) {
func (o *NewActorsOpts) registerCronJobs(host *local.Host) (err error) {
// In test mode, we do not register anything
if common.EnvConfig.AppEnv == "test" {
return opts, nil
return nil
}
// Register the analytics job
analyticsJob, err := job.GetAnalyticsJob(o.AppConfig, o.HttpClient)
analyticsJob, err := job.GetAnalyticsJob(o.HttpClient, o.InstanceID)
if err != nil {
return nil, fmt.Errorf("failed to get analytics cron job: %w", err)
return fmt.Errorf("failed to get analytics cron job: %w", err)
}
// This could be nil if analytics are disabled
if analyticsJob != nil {
// This could be nil if analytics are disabled
opts = append(opts, local.WithBuiltInActor(analyticsJob))
err = host.RegisterBuiltInActor(analyticsJob)
if err != nil {
return fmt.Errorf("error registering built-in actor for analytics job: %w", err)
}
}
// Register the file cleanup jobs
fileCleanupJobs, err := job.GetFileCleanupJobs(o.DB, o.FileStorage)
if err != nil {
return nil, fmt.Errorf("failed to get file cleanup cron jobs: %w", err)
return fmt.Errorf("failed to get file cleanup cron jobs: %w", err)
}
for _, j := range fileCleanupJobs {
opts = append(opts, local.WithBuiltInActor(j))
err = host.RegisterBuiltInActor(j)
if err != nil {
return fmt.Errorf("error registering built-in actor for cleanup job: %w", err)
}
}
return opts, nil
return nil
}
// getRateLimiters creates a built-in rate-limit actor for each middleware policy and returns both the created actors (keyed by policy name) and the host options to register them
// registerRateLimiters creates a built-in rate-limit actor for each middleware policy and returns both the created actors (keyed by policy name) and the host options to register them
// Unlike cron jobs, rate limiters keep no durable state, so they are registered in every environment
func (o *NewActorsOpts) getRateLimiters() (actors map[string]*ratelimit.RateLimit, opts []local.HostOption, err error) {
func (o *NewActorsOpts) registerRateLimiters(host *local.Host) (actors map[string]*ratelimit.RateLimit, err error) {
policies := middleware.RateLimitPolicies()
actors = make(map[string]*ratelimit.RateLimit, len(policies))
for _, p := range policies {
@@ -152,11 +157,15 @@ func (o *NewActorsOpts) getRateLimiters() (actors map[string]*ratelimit.RateLimi
ratelimit.WithBurst(p.Burst),
)
if err != nil {
return nil, nil, fmt.Errorf("error creating rate limiter %q: %w", p.Name, err)
return nil, fmt.Errorf("error creating rate limiter %q: %w", p.Name, err)
}
actors[p.Name] = rl
opts = append(opts, local.WithBuiltInActor(rl))
err = host.RegisterBuiltInActor(rl)
if err != nil {
return nil, fmt.Errorf("error registering built-in actor for rate limiter '%s': %w", p.Name, err)
}
}
return actors, opts, nil
return actors, nil
}
backend/internal/bootstrap/actors_bootstrap_test.go
+4 -2
View File
@@ -14,13 +14,15 @@ func TestNewActorsOptsGetPSKUsesStableValue(t *testing.T) {
EnvConfig: &common.EnvConfigSchema{
EncryptionKey: []byte("test-encryption-key"),
},
// Constant value for this test
InstanceID: "ee05c3eb-8129-47a6-a1c7-849998b6f876",
}
expectedHex := "651300d35d48998d0fa66ac89091bcde8ed0fd0aa35fbb849f068410c64807e9"
expectedHex := "db09067fa194c3731bf77b6415a1c5d903f03d4557605ba3236b31f6eddfc8d7"
expected, err := hex.DecodeString(expectedHex)
require.NoError(t, err)
actual, err := opts.getPSK()
require.NoError(t, err)
require.Equal(t, expected, actual)
require.Equalf(t, expected, actual, "actual result: %s", actual)
}
backend/internal/bootstrap/bootstrap.go
+70 -24
View File
@@ -9,10 +9,12 @@ import (
_ "github.com/golang-migrate/migrate/v4/source/file"
"github.com/italypaleale/francis/host/local"
"github.com/italypaleale/go-kit/servicerunner"
"gorm.io/gorm"
"github.com/pocket-id/pocket-id/backend/internal/common"
"github.com/pocket-id/pocket-id/backend/internal/instanceid"
"github.com/pocket-id/pocket-id/backend/internal/job"
"github.com/pocket-id/pocket-id/backend/internal/service"
"github.com/pocket-id/pocket-id/backend/internal/storage"
@@ -46,6 +48,13 @@ func Bootstrap(ctx context.Context) error {
}()
}
// Load the instance ID
// This is stored in the "kv" table, and generated on first startup
instanceID, err := instanceid.Load(ctx, db)
if err != nil {
return fmt.Errorf("failed to initialize instance ID: %w", err)
}
// Init storage
fileStorage, err := InitStorage(ctx, db)
if err != nil {
@@ -64,8 +73,34 @@ func Bootstrap(ctx context.Context) error {
return fmt.Errorf("failed to create job scheduler: %w", err)
}
// Init the actors
// The actor host is created and started before the services, so services can depend on it once it's ready
actorsOpts := NewActorsOpts{
Postgres: pg,
EnvConfig: &common.EnvConfig,
InstanceID: instanceID,
HttpClient: httpClient,
DB: db,
FileStorage: fileStorage,
}
if pg == nil {
actorsOpts.SQLite, err = db.DB()
if err != nil {
return fmt.Errorf("failed to get *sql.DB connection from Gorm: %w", err)
}
}
actors, rateLimitServices, err := NewActors(actorsOpts)
if err != nil {
return fmt.Errorf("failed to initialize actors: %w", err)
}
// Run the actor host as a background service and get a "ready" signal that other services can wait on
actorsRun, actorsReady := actorsRunServiceFn(actors)
services = append(services, actorsRun)
// Create all services
svc, err := initServices(ctx, db, httpClient, imageExtensions, fileStorage, scheduler)
svc, err := initServices(ctx, db, instanceID, httpClient, imageExtensions, fileStorage, scheduler)
if err != nil {
return fmt.Errorf("failed to initialize services: %w", err)
}
@@ -93,28 +128,6 @@ func Bootstrap(ctx context.Context) error {
return nil
})
// Init the actors
actorsOpts := NewActorsOpts{
Postgres: pg,
EnvConfig: &common.EnvConfig,
AppConfig: svc.appConfigService,
HttpClient: httpClient,
DB: db,
FileStorage: fileStorage,
}
if pg == nil {
actorsOpts.SQLite, err = db.DB()
if err != nil {
return fmt.Errorf("failed to get *sql.DB connection from Gorm: %w", err)
}
}
actors, rateLimitServices, err := NewActors(actorsOpts)
if err != nil {
return fmt.Errorf("failed to initialize actors: %w", err)
}
services = append(services, actors.Run)
// Register scheduled jobs, only in non-test mode
if common.EnvConfig.AppEnv != "test" {
err = registerScheduledJobs(ctx, db, svc, scheduler)
@@ -131,7 +144,9 @@ func Bootstrap(ctx context.Context) error {
if err != nil {
return fmt.Errorf("failed to initialize router: %w", err)
}
services = append(services, router)
// The router must wait on the actor host being ready, since the rate-limit middleware invokes actors
services = append(services, actorsReady.Await(router))
// Run all background services
// This call blocks until the context is canceled
@@ -146,6 +161,37 @@ func Bootstrap(ctx context.Context) error {
return nil
}
// actorsRunServiceFn wraps the actor host's Run method in a background service and returns a "ready" signal that other services can wait on
func actorsRunServiceFn(actors *local.Host) (servicerunner.Service, *servicerunner.Ready) {
actorsReady := servicerunner.NewReady()
fn := func(ctx context.Context) error {
runErrCh := make(chan error, 1)
go func() {
runErrCh <- actors.Run(ctx)
}()
// Wait for the right signal
select {
case <-actors.Ready():
// Actor host is ready, signal actorsReady
actorsReady.Signal()
case runErr := <-runErrCh:
// Run returned with an error
return runErr
case <-ctx.Done():
// Context canceled
return ctx.Err()
}
// Now the actor host is running
// This goroutine must stay up until the actor host returns
// Here, context cancellation will surface through this channel too
return <-runErrCh
}
return fn, actorsReady
}
func InitStorage(ctx context.Context, db *gorm.DB) (fileStorage storage.FileStorage, err error) {
switch common.EnvConfig.FileBackend {
case storage.TypeFileSystem:
backend/internal/bootstrap/services_bootstrap.go
+2 -2
View File
@@ -44,7 +44,7 @@ type services struct {
}
// Initializes all services
func initServices(ctx context.Context, db *gorm.DB, httpClient *http.Client, imageExtensions map[string]string, fileStorage storage.FileStorage, scheduler *job.Scheduler) (svc *services, err error) {
func initServices(ctx context.Context, db *gorm.DB, instanceID string, httpClient *http.Client, imageExtensions map[string]string, fileStorage storage.FileStorage, scheduler *job.Scheduler) (svc *services, err error) {
svc = &services{}
svc.appConfigService, err = service.NewAppConfigService(ctx, db)
@@ -63,7 +63,7 @@ func initServices(ctx context.Context, db *gorm.DB, httpClient *http.Client, ima
svc.geoLiteService = service.NewGeoLiteService(httpClient)
svc.auditLogService = service.NewAuditLogService(db, svc.appConfigService, svc.emailService, svc.geoLiteService)
svc.jwtService, err = service.NewJwtService(ctx, db, svc.appConfigService)
svc.jwtService, err = service.NewJwtService(ctx, db, instanceID, svc.appConfigService)
if err != nil {
return nil, fmt.Errorf("failed to create JWT service: %w", err)
}