Project Architecture

type ServiceParams struct {
    // Individual Repository Interfaces
    AuthRepo                     auth.Repository
	UserRepo                     user.Repository
	EventRepo                    events.Repository
	ProcessedEventRepo           events.ProcessedEventRepository
    // ... 40+ more domain specific repositories

    // Publishers
	EventPublisher   publisher.EventPublisher
	WebhookPublisher webhookPublisher.WebhookPublisher

    // Services
    Logger       *logger.Logger
	Config       *config.Configuration
	DB           postgres.IClient
	PDFGenerator pdf.Generator
	S3           s3.Service

    // http client
	Client httpclient.Client

	// Proration
	ProrationCalculator proration.Calculator

	// Integration Factory
	IntegrationFactory *integration.Factory
}

/v1
├── /auth                    # Authentication endpoints
├── /customers              # Customer management
├── /plans                  # Pricing plans and features
├── /subscriptions          # Subscription lifecycle
├── /events                 # Event ingestion
├── /invoices              # Invoice generation and retrieval
├── /usage                 # Usage queries and aggregations
├── /payments              # Payment processing
└── /webhooks              # Webhook management

func AuthenticateMiddleware(cfg *config.Configuration, secretService service.SecretService, logger *logger.Logger) gin.HandlerFunc {
    return func(c *gin.Context) {
        // Check for API key in X-API-Key header

        // Check for JWT token in Authorization header

        c.Next()
    }
}

type Repository interface {
    // All methods automatically filter by tenant/environment from context
    GetCustomer(ctx context.Context, customerID string) (*Customer, error)
    CreateInvoice(ctx context.Context, invoice *Invoice) error
    GetUsageEvents(ctx context.Context, filters EventFilters) ([]Event, error)
}

func (r *PostgresRepository) GetCustomer(ctx context.Context, customerID string) (*Customer, error) {
    tenantID := ctx.Value(types.CtxTenantID).(string)
    environment := ctx.Value(types.CtxEnvironmentID).(string)

    return r.db.Customer.
        Query().
        Where(customer.ID(customerID)).
        Where(customer.TenantID(tenantID)).
        Where(customer.Environment(environment)).
        Only(ctx)
}

router := message.NewRouter(message.RouterConfig{}, logger)

// Event processing pipeline
router.AddHandler(
    "events_processor",
    "events",
    subscriber,
    "events_post_processing",
    publisher,
    eventProcessingHandler,
)

func RegisterWorkflows(worker worker.Worker) {
    // Billing workflows
    worker.RegisterWorkflow(billing.GenerateInvoiceWorkflow)
    worker.RegisterWorkflow(billing.ProcessPaymentWorkflow)

    // Usage workflows
    worker.RegisterWorkflow(usage.AggregateUsageWorkflow)

    // Export workflows
    worker.RegisterWorkflow(export.ExportDataWorkflow)
}

func StartTemporalWorker(config *config.Config) {
    client := temporal.NewClient()
    worker := worker.New(client, "billing-queue", worker.Options{})

    RegisterWorkflows(worker)
    RegisterActivities(worker)

    worker.Run(worker.InterruptCh())
}

// Integration Factory creates specific integration instances
type Factory struct {
    config *config.Configuration
    logger *logger.Logger
}

// Specific integration getters with proper error handling
func (f *Factory) GetStripeIntegration(ctx context.Context) (*StripeIntegration, error) {
    // Initialize and return Stripe integration with configuration
    return NewStripeIntegration(f.config, f.logger), nil
}

func (f *Factory) GetHubSpotIntegration(ctx context.Context) (*HubSpotIntegration, error) {
    // Initialize and return HubSpot integration with configuration
    return NewHubSpotIntegration(f.config, f.logger), nil
}

func (f *Factory) GetRazorpayIntegration(ctx context.Context) (*RazorpayIntegration, error) {
    // Initialize and return Razorpay integration with configuration
    return NewRazorpayIntegration(f.config, f.logger), nil
}

type WebhookService struct {
    config    *config.Configuration
    publisher publisher.WebhookPublisher
    handler   handler.Handler
    factory   payload.PayloadBuilderFactory
    client    httpclient.Client
    logger    *logger.Logger
}

func (w *WebhookService) DeliverEvent(event *Event, endpoints []WebhookEndpoint) error {
    // Route based on configuration
    if w.config.Webhook.SvixConfig.Enabled {
        w.logger.Info("Using Svix for webhook delivery")
        return w.deliverViaSvix(event)
    }

    w.logger.Info("Using internal webhook system")
    return w.deliverViaInternalSystem(event, endpoints)
}

// Configuration controls webhook delivery method
// svix_config:
//   enabled: false  # Default - uses internal system
//   enabled: true   # Optional - uses Svix for delivery

sentry.Init(sentry.ClientOptions{
    Dsn: config.SentryDSN,
    Environment: config.Environment,
    TracesSampleRate: 0.1,
})

// Error tracking in handlers
if err != nil {
    sentry.CaptureException(err)
    return gin.Error{Err: err, Type: gin.ErrorTypePublic}
}

pyroscope.Start(pyroscope.Config{
    ApplicationName: "flexprice-api",
    ServerAddress:   config.PyroscopeURL,
    ProfileTypes: []pyroscope.ProfileType{
        pyroscope.ProfileCPU,
        pyroscope.ProfileAllocObjects,
        pyroscope.ProfileAllocSpace,
    },
})

func (s *S3Service) StoreInvoice(invoiceID string, pdfData []byte) (string, error) {
    key := fmt.Sprintf("invoices/%s/%s.pdf", s.tenantID, invoiceID)

    _, err := s.client.PutObject(&s3.PutObjectInput{
        Bucket: aws.String(s.bucket),
        Key:    aws.String(key),
        Body:   bytes.NewReader(pdfData),
    })

    // Return presigned URL for download
    return s.GeneratePresignedURL(key, 24*time.Hour)
}

// Cache interface in internal/cache/
type Cache interface {
    // Get retrieves a value from the cache
    // Returns the value and a boolean indicating whether the key was found
    Get(ctx context.Context, key string) (interface{}, bool)

    // Set adds a value to the cache with the specified expiration
    // If expiration is 0, the item never expires (but may be evicted)
    Set(ctx context.Context, key string, value interface{}, expiration time.Duration)

    // Delete removes a key from the cache
    Delete(ctx context.Context, key string)

    // DeleteByPrefix removes all keys with the given prefix
    DeleteByPrefix(ctx context.Context, prefix string)

    // Flush removes all items from the cache
    Flush(ctx context.Context)
}

// Usage example
func (s *PlanService) GetPlan(ctx context.Context, planID string) (*Plan, error) {
    // Check if caching is enabled
    if !s.config.Cache.Enabled {
        return s.repo.GetPlan(ctx, planID)
    }

    // Try cache first
    if cached, found := s.cache.Get(ctx, planID); found {
        return cached.(*Plan), nil
    }

    // Cache miss - fetch from database
    plan, err := s.repo.GetPlan(ctx, planID)
    if err == nil {
        s.cache.Set(ctx, planID, plan, time.Hour)
    }
    return plan, err
}

// Configuration
// cache:
//   enabled: true/false  # Simple boolean to enable/disable caching

# config.yaml
server:
  port: 8080
  host: "0.0.0.0"

database:
  postgres:
    host: ${DB_HOST:localhost}
    port: ${DB_PORT:5432}
    database: ${DB_NAME:flexprice}
  clickhouse:
    host: ${CLICKHOUSE_HOST:localhost}
    port: ${CLICKHOUSE_PORT:9000}

kafka:
  brokers: ${KAFKA_BROKERS:localhost:29092}
  topics:
    events: "events"
    events_lazy: "events_lazy"
    processed_events: "events_post_processing"
    system_events: "system_events"

temporal:
  host: ${TEMPORAL_HOST:localhost:7233}
  namespace: ${TEMPORAL_NAMESPACE:default}

type Configuration struct {
    Deployment               DeploymentConfig
    Server                   ServerConfig
    Auth                     AuthConfig
    Kafka                    KafkaConfig
    ClickHouse               ClickHouseConfig
    Logging                  LoggingConfig
    Postgres                 PostgresConfig
    Sentry                   SentryConfig
    Pyroscope                PyroscopeConfig
    Event                    EventConfig
    DynamoDB                 DynamoDBConfig
    Temporal                 TemporalConfig
    Webhook                  Webhook
    Secrets                  SecretsConfig
    Billing                  BillingConfig
    S3                       S3Config
    Cache                    CacheConfig
    EventProcessing          EventProcessingConfig
    EventProcessingLazy      EventProcessingLazyConfig
    EventPostProcessing      EventPostProcessingConfig
    FeatureUsageTracking     FeatureUsageTrackingConfig
    FeatureUsageTrackingLazy FeatureUsageTrackingLazyConfig
    EnvAccess                EnvAccessConfig
    FeatureFlag              FeatureFlagConfig
    Email                    EmailConfig
}

func main() {
    fx.New(
        // Configuration
        fx.Provide(config.Load),
        fx.Provide(logger.New),

        // Infrastructure
        fx.Provide(database.NewPostgres),
        fx.Provide(database.NewClickHouse),
        fx.Provide(kafka.NewProducer),
        fx.Provide(temporal.NewClient),

        // Services
        fx.Provide(service.NewBillingService),
        fx.Provide(service.NewEventService),

        // HTTP Server
        fx.Provide(server.New),
        fx.Invoke(server.RegisterRoutes),

        // Lifecycle hooks
        fx.Invoke(func(lc fx.Lifecycle, server *server.Server) {
            lc.Append(fx.Hook{
                OnStart: server.Start,
                OnStop:  server.Stop,
            })
        }),
    ).Run()
}