diff --git a/internal/api/automations/handler.go b/internal/api/automations/handler.go index 09a7a51..4dc60cd 100644 --- a/internal/api/automations/handler.go +++ b/internal/api/automations/handler.go @@ -764,12 +764,14 @@ func (h *Handler) convertAutomationToMapWithDeviceSpace(ctx context.Context, a * // Helper functions to convert models to maps func convertAutomationToMap(a *models.AutomationWithActions) map[string]interface{} { result := map[string]interface{}{ - "id": a.ID, - "name": a.Name, - "title": a.Title, - "device_id": a.DeviceID, - "updated_at": a.UpdatedAt, - "created_at": a.CreatedAt, + "id": a.ID, + "name": a.Name, + "title": a.Title, + "device_id": a.DeviceID, + "is_deactivated": a.IsDeactivated, + "deactivated_at": a.DeactivatedAt, + "updated_at": a.UpdatedAt, + "created_at": a.CreatedAt, } if a.EventRule != nil { diff --git a/internal/celery/consumer.go b/internal/celery/consumer.go index 286fe3a..98e6020 100644 --- a/internal/celery/consumer.go +++ b/internal/celery/consumer.go @@ -21,12 +21,16 @@ const ( DeleteSpaceExchange = "delete_space" DeleteDeviceExchange = "delete_device" CreateDeviceEntitiesExchange = "create_device_entities" + AutomationDowngradeExchange = "automation_downgrade" + AutomationUpgradeExchange = "automation_upgrade" // Task names for message identification UpdateSpaceTaskName = "spacedf.tasks.update_space" DeleteSpaceTaskName = "spacedf.tasks.delete_space" DeleteDeviceTaskName = "spacedf.tasks.delete_device" CreateDeviceEntitiesTaskName = "spacedf.tasks.create_device_entities" + AutomationDowngradeTaskName = "spacedf.tasks.automation_downgrade" + AutomationUpgradeTaskName = "spacedf.tasks.automation_upgrade" ) // TaskConsumer consumes Celery tasks from RabbitMQ @@ -45,6 +49,8 @@ type TaskConsumer struct { deleteQueueName string deviceQueueName string createEntitiesQueueName string + downgradeQueueName string + upgradeQueueName string } // SchemaInitializer handles database schema initialization @@ -64,6 +70,8 @@ func NewTaskConsumer(amqpURL string, dbClient *timescaledb.Client, logger *zap.L deleteQueueName: "telemetry_delete_space", deviceQueueName: "telemetry_delete_device", createEntitiesQueueName: "telemetry_create_device_entities", + downgradeQueueName: "telemetry_automation_downgrade", + upgradeQueueName: "telemetry_automation_upgrade", } } @@ -165,6 +173,42 @@ func (c *TaskConsumer) Connect() error { return fmt.Errorf("failed to declare create_device_entities exchange: %w", err) } + // Exchange: automation_downgrade (direct type) + err = c.channel.ExchangeDeclare( + AutomationDowngradeExchange, + "direct", + true, + false, + false, + true, + nil, + ) + if err != nil { + defer func() { + _ = c.channel.Close() + _ = c.conn.Close() + }() + return fmt.Errorf("failed to declare automation_downgrade exchange: %w", err) + } + + // Exchange: automation_upgrade (direct type) + err = c.channel.ExchangeDeclare( + AutomationUpgradeExchange, + "direct", + true, + false, + false, + true, + nil, + ) + if err != nil { + defer func() { + _ = c.channel.Close() + _ = c.conn.Close() + }() + return fmt.Errorf("failed to declare automation_upgrade exchange: %w", err) + } + // Declare update queue _, err = c.channel.QueueDeclare( c.updateQueueName, @@ -291,11 +335,65 @@ func (c *TaskConsumer) Connect() error { return fmt.Errorf("failed to bind create entities queue: %w", err) } + // Declare automation downgrade queue + _, err = c.channel.QueueDeclare( + c.downgradeQueueName, + true, + false, + false, + false, + amqp.Table{"x-single-active-consumer": true}, + ) + if err != nil { + _ = c.channel.Close() + _ = c.conn.Close() + return fmt.Errorf("failed to declare automation downgrade queue: %w", err) + } + if err := c.channel.QueueBind( + c.downgradeQueueName, + AutomationDowngradeExchange, + AutomationDowngradeTaskName, + false, + nil, + ); err != nil { + _ = c.channel.Close() + _ = c.conn.Close() + return fmt.Errorf("failed to bind automation downgrade queue: %w", err) + } + + // Declare automation upgrade queue + _, err = c.channel.QueueDeclare( + c.upgradeQueueName, + true, + false, + false, + false, + amqp.Table{"x-single-active-consumer": true}, + ) + if err != nil { + _ = c.channel.Close() + _ = c.conn.Close() + return fmt.Errorf("failed to declare automation upgrade queue: %w", err) + } + if err := c.channel.QueueBind( + c.upgradeQueueName, + AutomationUpgradeExchange, + AutomationUpgradeTaskName, + false, + nil, + ); err != nil { + _ = c.channel.Close() + _ = c.conn.Close() + return fmt.Errorf("failed to bind automation upgrade queue: %w", err) + } + c.logger.Info("Celery task consumer connected", zap.String("update_queue", c.updateQueueName), zap.String("delete_queue", c.deleteQueueName), zap.String("device_queue", c.deviceQueueName), - zap.String("create_entities_queue", c.createEntitiesQueueName)) + zap.String("create_entities_queue", c.createEntitiesQueueName), + zap.String("downgrade_queue", c.downgradeQueueName), + zap.String("upgrade_queue", c.upgradeQueueName)) return nil } @@ -403,6 +501,34 @@ func (c *TaskConsumer) connectAndConsume(ctx context.Context) error { return fmt.Errorf("failed to start consuming create entities queue: %w", err) } + // Consume from automation downgrade queue + downgradeMessages, err := c.channel.Consume( + c.downgradeQueueName, + "telemetry_downgrade_consumer", + false, + false, + false, + false, + nil, + ) + if err != nil { + return fmt.Errorf("failed to start consuming downgrade queue: %w", err) + } + + // Consume from automation upgrade queue + upgradeMessages, err := c.channel.Consume( + c.upgradeQueueName, + "telemetry_upgrade_consumer", + false, + false, + false, + false, + nil, + ) + if err != nil { + return fmt.Errorf("failed to start consuming upgrade queue: %w", err) + } + c.logger.Info("Celery task consumer started", zap.String("update_queue", c.updateQueueName), zap.String("delete_queue", c.deleteQueueName), @@ -410,7 +536,7 @@ func (c *TaskConsumer) connectAndConsume(ctx context.Context) error { zap.String("create_entities_queue", c.createEntitiesQueueName)) // Start goroutines for each queue - c.wg.Add(4) + c.wg.Add(6) go func() { defer c.wg.Done() c.processMessages(ctx, updateMessages, UpdateSpaceTaskName) @@ -427,6 +553,14 @@ func (c *TaskConsumer) connectAndConsume(ctx context.Context) error { defer c.wg.Done() c.processMessages(ctx, createEntitiesMessages, CreateDeviceEntitiesTaskName) }() + go func() { + defer c.wg.Done() + c.processMessages(ctx, downgradeMessages, AutomationDowngradeTaskName) + }() + go func() { + defer c.wg.Done() + c.processMessages(ctx, upgradeMessages, AutomationUpgradeTaskName) + }() // Wait for all goroutines to finish (they exit when channel closes) c.wg.Wait() @@ -499,6 +633,12 @@ func (c *TaskConsumer) handleTask(ctx context.Context, taskName string, body []b case CreateDeviceEntitiesTaskName, "create_device_entities": return c.handleCreateDeviceEntities(ctx, body) + case AutomationDowngradeTaskName, "automation_downgrade": + return c.handleAutomationDowngrade(ctx, body) + + case AutomationUpgradeTaskName, "automation_upgrade": + return c.handleAutomationUpgrade(ctx, body) + default: c.logger.Debug("Unknown task name, ignoring", zap.String("task", taskName)) return nil @@ -651,6 +791,64 @@ func (c *TaskConsumer) handleCreateDeviceEntities(ctx context.Context, body []by return nil } +func (c *TaskConsumer) handleAutomationDowngrade(ctx context.Context, body []byte) error { + var celeryMsg models.CeleryMessage + if err := json.Unmarshal(body, &celeryMsg); err != nil { + return fmt.Errorf("failed to unmarshal celery message: %w", err) + } + + var task models.AutomationDowngradeTask + if err := json.Unmarshal(celeryMsg.Kwargs, &task); err != nil { + return fmt.Errorf("failed to unmarshal automation_downgrade task kwargs: %w", err) + } + + var maxActive int + if task.Limits != nil { + if v, ok := task.Limits["automation.max_count"]; ok { + maxActive = v + } + } + + c.logger.Info("Processing automation downgrade", + zap.String("org", task.OrgSlug), + zap.Int("max_active", maxActive)) + + deactivated, err := c.dbClient.BulkDeactivateAutomations(ctx, task.OrgSlug, maxActive) + if err != nil { + return fmt.Errorf("failed to bulk deactivate automations: %w", err) + } + + c.logger.Info("Automation downgrade completed", + zap.String("org", task.OrgSlug), + zap.Int64("deactivated", deactivated)) + return nil +} + +func (c *TaskConsumer) handleAutomationUpgrade(ctx context.Context, body []byte) error { + var celeryMsg models.CeleryMessage + if err := json.Unmarshal(body, &celeryMsg); err != nil { + return fmt.Errorf("failed to unmarshal celery message: %w", err) + } + + var task models.AutomationUpgradeTask + if err := json.Unmarshal(celeryMsg.Kwargs, &task); err != nil { + return fmt.Errorf("failed to unmarshal automation_upgrade task kwargs: %w", err) + } + + c.logger.Info("Processing automation upgrade", + zap.String("org", task.OrgSlug)) + + reactivated, err := c.dbClient.BulkReactivateAutomations(ctx, task.OrgSlug) + if err != nil { + return fmt.Errorf("failed to bulk reactivate automations: %w", err) + } + + c.logger.Info("Automation upgrade completed", + zap.String("org", task.OrgSlug), + zap.Int64("reactivated", reactivated)) + return nil +} + // Stop gracefully stops the consumer func (c *TaskConsumer) Stop() error { c.stopOnce.Do(func() { diff --git a/internal/models/automations.go b/internal/models/automations.go index 737225f..4860004 100644 --- a/internal/models/automations.go +++ b/internal/models/automations.go @@ -17,15 +17,17 @@ type Action struct { // Automation represents an automation database row type Automation struct { - ID string - Name string - Title *string - DeviceID string - EventRuleID *string - EventRule *EventRule - SpaceID *uuid.UUID - UpdatedAt time.Time - CreatedAt time.Time + ID string + Name string + Title *string + DeviceID string + EventRuleID *string + EventRule *EventRule + SpaceID *uuid.UUID + IsDeactivated bool + DeactivatedAt *time.Time + UpdatedAt time.Time + CreatedAt time.Time } // AutomationWithActions represents an automation with its associated actions diff --git a/internal/models/celery.go b/internal/models/celery.go index 671f17a..4d4ebe4 100644 --- a/internal/models/celery.go +++ b/internal/models/celery.go @@ -152,3 +152,14 @@ type CreateDeviceEntitiesTask struct { DeviceModel string `json:"device_model"` DevEUI string `json:"dev_eui"` } + +// AutomationDowngradeTask represents the Celery task kwargs for automation_downgrade +type AutomationDowngradeTask struct { + OrgSlug string `json:"org_slug"` + Limits map[string]int `json:"limits"` +} + +// AutomationUpgradeTask represents the Celery task kwargs for automation_upgrade +type AutomationUpgradeTask struct { + OrgSlug string `json:"org_slug"` +} diff --git a/internal/timescaledb/automations.go b/internal/timescaledb/automations.go index 9c81080..d69f0fe 100644 --- a/internal/timescaledb/automations.go +++ b/internal/timescaledb/automations.go @@ -19,14 +19,16 @@ import ( // automationRow represents a database row exactly as returned by SQL // Using sql.Null* types directly eliminates manual null checking type automationRow struct { - ID sql.NullString `db:"id"` - Name sql.NullString `db:"name"` - Title sql.NullString `db:"title"` - DeviceID sql.NullString `db:"device_id"` - EventRuleID sql.NullString `db:"event_rule_id"` - SpaceID sql.NullString `db:"space_id"` - UpdatedAt sql.NullTime `db:"updated_at"` - CreatedAt sql.NullTime `db:"created_at"` + ID sql.NullString `db:"id"` + Name sql.NullString `db:"name"` + Title sql.NullString `db:"title"` + DeviceID sql.NullString `db:"device_id"` + EventRuleID sql.NullString `db:"event_rule_id"` + SpaceID sql.NullString `db:"space_id"` + UpdatedAt sql.NullTime `db:"updated_at"` + CreatedAt sql.NullTime `db:"created_at"` + IsDeactivated sql.NullBool `db:"is_deactivated"` + DeactivatedAt sql.NullTime `db:"deactivated_at"` // Event rule fields EREventRuleID sql.NullString `db:"er_event_rule_id"` ERRuleKey sql.NullString `db:"er_rule_key"` @@ -52,13 +54,15 @@ func nullPtr[T any](v T, valid bool) *T { // Returns an error if JSON parsing fails for actions or event rule definition. func (r *automationRow) toModel() (*models.AutomationWithActions, error) { a := models.Automation{ - ID: r.ID.String, - Name: r.Name.String, - DeviceID: r.DeviceID.String, - UpdatedAt: r.UpdatedAt.Time, - CreatedAt: r.CreatedAt.Time, - Title: nullPtr(r.Title.String, r.Title.Valid), - EventRuleID: nullPtr(r.EventRuleID.String, r.EventRuleID.Valid), + ID: r.ID.String, + Name: r.Name.String, + DeviceID: r.DeviceID.String, + IsDeactivated: r.IsDeactivated.Bool, + DeactivatedAt: nullPtr(r.DeactivatedAt.Time, r.DeactivatedAt.Valid), + UpdatedAt: r.UpdatedAt.Time, + CreatedAt: r.CreatedAt.Time, + Title: nullPtr(r.Title.String, r.Title.Valid), + EventRuleID: nullPtr(r.EventRuleID.String, r.EventRuleID.Valid), } // Parse SpaceID as UUID @@ -143,7 +147,7 @@ func (c *Client) GetAutomations(ctx context.Context, spaceID uuid.UUID, deviceID // Query automations with actions query := ` SELECT a.id, a.name, a.title, a.device_id, - a.event_rule_id, a.space_id, a.updated_at, a.created_at, + a.event_rule_id, a.is_deactivated, a.deactivated_at, a.space_id, a.updated_at, a.created_at, er.event_rule_id, er.rule_key, er.definition, er.is_active, er.repeat_able, er.cooldown_sec, er.description, COALESCE( json_agg( @@ -162,7 +166,7 @@ func (c *Client) GetAutomations(ctx context.Context, spaceID uuid.UUID, deviceID LEFT JOIN actions act ON act.id = aa.action_id LEFT JOIN event_rules er ON er.event_rule_id = a.event_rule_id ` + whereClause + ` - GROUP BY a.id, a.name, a.title, a.device_id, a.event_rule_id, a.space_id, a.updated_at, a.created_at, er.event_rule_id, er.rule_key, er.definition::text, er.is_active, er.repeat_able, er.cooldown_sec, er.description + GROUP BY a.id, a.name, a.title, a.device_id, a.event_rule_id, a.is_deactivated, a.deactivated_at, a.space_id, a.updated_at, a.created_at, er.event_rule_id, er.rule_key, er.definition::text, er.is_active, er.repeat_able, er.cooldown_sec, er.description ORDER BY a.created_at DESC LIMIT $` + fmt.Sprint(qb.argIndex) + ` OFFSET $` + fmt.Sprint(qb.argIndex+1) qb.AddLimitOffset(limit, offset) @@ -178,7 +182,7 @@ func (c *Client) GetAutomations(ctx context.Context, spaceID uuid.UUID, deviceID var row automationRow if err := rows.Scan( &row.ID, &row.Name, &row.Title, &row.DeviceID, - &row.EventRuleID, &row.SpaceID, &row.UpdatedAt, &row.CreatedAt, + &row.EventRuleID, &row.IsDeactivated, &row.DeactivatedAt, &row.SpaceID, &row.UpdatedAt, &row.CreatedAt, &row.EREventRuleID, &row.ERRuleKey, &row.ERDefinition, &row.ERIsActive, &row.ERRepeatAble, &row.ERCooldownSec, &row.ERDescription, &row.ActionsJSON, ); err != nil { @@ -219,7 +223,7 @@ func (c *Client) GetAutomationByID(ctx context.Context, automationID string) (*m err := c.WithOrgTx(ctx, org, func(txCtx context.Context, tx bob.Tx) error { query := ` SELECT a.id, a.name, a.title, a.device_id, - a.event_rule_id, a.space_id, a.updated_at, a.created_at, + a.event_rule_id, a.is_deactivated, a.space_id, a.updated_at, a.created_at, er.event_rule_id, er.rule_key, er.definition, er.is_active, er.repeat_able, er.cooldown_sec, er.description, COALESCE( json_agg( @@ -238,13 +242,13 @@ func (c *Client) GetAutomationByID(ctx context.Context, automationID string) (*m LEFT JOIN actions act ON act.id = aa.action_id LEFT JOIN event_rules er ON er.event_rule_id = a.event_rule_id WHERE a.id = $1 - GROUP BY a.id, a.name, a.title, a.device_id, a.event_rule_id, a.space_id, a.updated_at, a.created_at, er.event_rule_id, er.rule_key, er.definition::text, er.is_active, er.repeat_able, er.cooldown_sec, er.description + GROUP BY a.id, a.name, a.title, a.device_id, a.event_rule_id, a.is_deactivated, a.deactivated_at, a.space_id, a.updated_at, a.created_at, er.event_rule_id, er.rule_key, er.definition::text, er.is_active, er.repeat_able, er.cooldown_sec, er.description ` var row automationRow err := tx.QueryRowContext(txCtx, query, automationID).Scan( &row.ID, &row.Name, &row.Title, &row.DeviceID, - &row.EventRuleID, &row.SpaceID, &row.UpdatedAt, &row.CreatedAt, + &row.EventRuleID, &row.IsDeactivated, &row.DeactivatedAt, &row.SpaceID, &row.UpdatedAt, &row.CreatedAt, &row.EREventRuleID, &row.ERRuleKey, &row.ERDefinition, &row.ERIsActive, &row.ERRepeatAble, &row.ERCooldownSec, &row.ERDescription, &row.ActionsJSON, ) @@ -900,3 +904,47 @@ func (c *Client) GetAutomationSummary(ctx context.Context, spaceID uuid.UUID) (* return &stats, nil } + +func (c *Client) BulkDeactivateAutomations(ctx context.Context, org string, maxActive int) (int64, error) { + var deactivated int64 + err := c.WithOrgTx(ctx, org, func(txCtx context.Context, tx bob.Tx) error { + query := ` + UPDATE automations + SET is_deactivated = true, + deactivated_at = NOW() + WHERE id IN ( + SELECT id FROM automations + WHERE is_deactivated = false + ORDER BY created_at ASC + OFFSET $1 + ) + ` + result, err := tx.ExecContext(txCtx, query, maxActive) + if err != nil { + return err + } + deactivated, err = result.RowsAffected() + return err + }) + if err != nil { + return 0, fmt.Errorf("failed to bulk deactivate automations: %w", err) + } + return deactivated, nil +} + +func (c *Client) BulkReactivateAutomations(ctx context.Context, org string) (int64, error) { + var reactivated int64 + err := c.WithOrgTx(ctx, org, func(txCtx context.Context, tx bob.Tx) error { + query := `UPDATE automations SET is_deactivated = false, deactivated_at = NULL WHERE is_deactivated = true` + result, err := tx.ExecContext(txCtx, query) + if err != nil { + return err + } + reactivated, err = result.RowsAffected() + return err + }) + if err != nil { + return 0, fmt.Errorf("failed to bulk reactivate automations: %w", err) + } + return reactivated, nil +} diff --git a/pkgs/db/migrations/20260724000000_add_is_deactivated_to_automations.sql b/pkgs/db/migrations/20260724000000_add_is_deactivated_to_automations.sql new file mode 100644 index 0000000..e206b0f --- /dev/null +++ b/pkgs/db/migrations/20260724000000_add_is_deactivated_to_automations.sql @@ -0,0 +1,8 @@ +-- migrate:up +ALTER TABLE automations ADD COLUMN IF NOT EXISTS is_deactivated BOOLEAN DEFAULT FALSE; +ALTER TABLE automations ADD COLUMN IF NOT EXISTS deactivated_at TIMESTAMP WITH TIME ZONE; + +-- migrate:down + +ALTER TABLE automations DROP COLUMN IF EXISTS is_deactivated; +ALTER TABLE automations DROP COLUMN IF EXISTS deactivated_at;