Use sqlx.Tx for event processing

Author: yhabteab

internal/event/event.go

@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"fmt"
 	"github.com/icinga/icinga-notifications/internal/utils"
-	"github.com/icinga/icingadb/pkg/icingadb"
 	"github.com/icinga/icingadb/pkg/types"
 	"time"
 )
@@ -73,30 +72,6 @@ func (e *Event) FullString() string {
 	return b.String()
 }
 
-// Sync transforms this event to *event.EventRow and calls its sync method.
-func (e *Event) Sync(db *icingadb.DB, objectId types.Binary) error {
-	if e.ID != 0 {
-		return nil
-	}
-
-	eventRow := &EventRow{
-		Time:     types.UnixMilli(e.Time),
-		SourceID: e.SourceId,
-		ObjectID: objectId,
-		Type:     utils.ToDBString(e.Type),
-		Severity: e.Severity,
-		Username: utils.ToDBString(e.Username),
-		Message:  utils.ToDBString(e.Message),
-	}
-
-	err := eventRow.Sync(db)
-	if err == nil {
-		e.ID = eventRow.ID
-	}
-
-	return err
-}
-
 // EventRow represents a single event database row and isn't an in-memory representation of an event.
 type EventRow struct {
 	ID       int64           `db:"id"`
@@ -114,15 +89,14 @@ func (er *EventRow) TableName() string {
 	return "event"
 }
 
-// Sync synchronizes this types data to the database.
-// Returns an error when any of the database operation fails.
-func (er *EventRow) Sync(db *icingadb.DB) error {
-	eventId, err := utils.InsertAndFetchId(db, utils.BuildInsertStmtWithout(db, er, "id"), er)
-	if err != nil {
-		return err
+func NewEventRow(e Event, objectId types.Binary) *EventRow {
+	return &EventRow{
+		Time:     types.UnixMilli(e.Time),
+		SourceID: e.SourceId,
+		ObjectID: objectId,
+		Type:     utils.ToDBString(e.Type),
+		Severity: e.Severity,
+		Username: utils.ToDBString(e.Username),
+		Message:  utils.ToDBString(e.Message),
 	}
-
-	er.ID = eventId
-
-	return nil
 }

internal/incident/db_types.go

@@ -7,6 +7,7 @@ import (
 	"github.com/icinga/icinga-notifications/internal/utils"
 	"github.com/icinga/icingadb/pkg/icingadb"
 	"github.com/icinga/icingadb/pkg/types"
+	"github.com/jmoiron/sqlx"
 )
 
 type IncidentRow struct {
@@ -32,15 +33,15 @@ func (i *IncidentRow) Upsert() interface{} {
 // Sync synchronizes incidents to the database.
 // Fetches the last inserted incident id and modifies this incident's id.
 // Returns an error on database failure.
-func (i *IncidentRow) Sync(db *icingadb.DB, upsert bool) error {
+func (i *IncidentRow) Sync(tx *sqlx.Tx, db *icingadb.DB, upsert bool) error {
 	if upsert {
 		stmt, _ := db.BuildUpsertStmt(i)
-		_, err := db.NamedExec(stmt, i)
+		_, err := tx.NamedExec(stmt, i)
 		if err != nil {
 			return fmt.Errorf("failed to upsert incident: %s", err)
 		}
 	} else {
-		incidentId, err := utils.InsertAndFetchId(db, utils.BuildInsertStmtWithout(db, i, "id"), i)
+		incidentId, err := utils.InsertAndFetchId(tx, utils.BuildInsertStmtWithout(db, i, "id"), i)
 		if err != nil {
 			return err
 		}

internal/incident/history.go

@@ -7,13 +7,14 @@ import (
 	"github.com/icinga/icinga-notifications/internal/rule"
 	"github.com/icinga/icinga-notifications/internal/utils"
 	"github.com/icinga/icingadb/pkg/types"
+	"github.com/jmoiron/sqlx"
 	"time"
 )
 
 // Sync initiates an *incident.IncidentRow from the current incident state and syncs it with the database.
 // Before syncing any incident related database entries, this method should be called at least once.
 // Returns an error on db failure.
-func (i *Incident) Sync() error {
+func (i *Incident) Sync(tx *sqlx.Tx) error {
 	incidentRow := &IncidentRow{
 		ID:          i.incidentRowID,
 		ObjectID:    i.Object.ID,
@@ -22,7 +23,7 @@ func (i *Incident) Sync() error {
 		Severity:    i.Severity(),
 	}
 
-	err := incidentRow.Sync(i.db, i.incidentRowID != 0)
+	err := incidentRow.Sync(tx, i.db, i.incidentRowID != 0)
 	if err != nil {
 		return err
 	}
@@ -32,54 +33,51 @@ func (i *Incident) Sync() error {
 	return nil
 }
 
-func (i *Incident) AddHistory(historyRow *HistoryRow, fetchId bool) (types.Int, error) {
+func (i *Incident) AddHistory(tx *sqlx.Tx, historyRow *HistoryRow, fetchId bool) (types.Int, error) {
 	historyRow.IncidentID = i.incidentRowID
 
 	stmt := utils.BuildInsertStmtWithout(i.db, historyRow, "id")
 	if fetchId {
-		historyId, err := utils.InsertAndFetchId(i.db, stmt, historyRow)
+		historyId, err := utils.InsertAndFetchId(tx, stmt, historyRow)
 		if err != nil {
 			return types.Int{}, err
 		}
 
 		return utils.ToDBInt(historyId), nil
 	} else {
-		_, err := i.db.NamedExec(stmt, historyRow)
+		_, err := tx.NamedExec(stmt, historyRow)
 		if err != nil {
-			return types.Int{}, fmt.Errorf("failed to insert incident history: %w", err)
+			return types.Int{}, err
 		}
 	}
 
 	return types.Int{}, nil
 }
 
-func (i *Incident) AddEscalationTriggered(state *EscalationState, hr *HistoryRow) (types.Int, error) {
+func (i *Incident) AddEscalationTriggered(tx *sqlx.Tx, state *EscalationState, hr *HistoryRow) (types.Int, error) {
 	state.IncidentID = i.incidentRowID
 
 	stmt, _ := i.db.BuildUpsertStmt(state)
-	_, err := i.db.NamedExec(stmt, state)
+	_, err := tx.NamedExec(stmt, state)
 	if err != nil {
-		return types.Int{}, fmt.Errorf("failed to insert incident rule escalation state: %w", err)
+		return types.Int{}, err
 	}
 
-	return i.AddHistory(hr, true)
+	return i.AddHistory(tx, hr, true)
 }
 
 // AddEvent Inserts incident history record to the database and returns an error on db failure.
-func (i *Incident) AddEvent(ev *event.Event) error {
+func (i *Incident) AddEvent(tx *sqlx.Tx, ev *event.Event) error {
 	ie := &EventRow{IncidentID: i.incidentRowID, EventID: ev.ID}
 	stmt, _ := i.db.BuildInsertStmt(ie)
-	_, err := i.db.NamedExec(stmt, ie)
-	if err != nil {
-		return fmt.Errorf("failed to insert incident event: %w", err)
-	}
+	_, err := tx.NamedExec(stmt, ie)
 
-	return nil
+	return err
 }
 
 // AddRecipient adds recipient from the given *rule.Escalation to this incident.
 // Syncs also all the recipients with the database and returns an error on db failure.
-func (i *Incident) AddRecipient(escalation *rule.Escalation, eventId int64) error {
+func (i *Incident) AddRecipient(tx *sqlx.Tx, escalation *rule.Escalation, eventId int64) error {
 	newRole := RoleRecipient
 	if i.HasManager() {
 		newRole = RoleSubscriber
@@ -112,7 +110,7 @@ func (i *Incident) AddRecipient(escalation *rule.Escalation, eventId int64) erro
 					OldRecipientRole: oldRole,
 				}
 
-				_, err := i.AddHistory(hr, false)
+				_, err := i.AddHistory(tx, hr, false)
 				if err != nil {
 					return err
 				}
@@ -121,7 +119,7 @@ func (i *Incident) AddRecipient(escalation *rule.Escalation, eventId int64) erro
 		}
 
 		stmt, _ := i.db.BuildUpsertStmt(cr)
-		_, err := i.db.NamedExec(stmt, cr)
+		_, err := tx.NamedExec(stmt, cr)
 		if err != nil {
 			return fmt.Errorf("failed to upsert incident contact %s: %w", r, err)
 		}
@@ -132,18 +130,18 @@ func (i *Incident) AddRecipient(escalation *rule.Escalation, eventId int64) erro
 
 // AddRuleMatchedHistory syncs the given *rule.Rule and history entry to the database.
 // Returns an error on database failure.
-func (i *Incident) AddRuleMatchedHistory(r *rule.Rule, hr *HistoryRow) (types.Int, error) {
+func (i *Incident) AddRuleMatchedHistory(tx *sqlx.Tx, r *rule.Rule, hr *HistoryRow) (types.Int, error) {
 	rr := &RuleRow{IncidentID: i.incidentRowID, RuleID: r.ID}
 	stmt, _ := i.db.BuildUpsertStmt(rr)
-	_, err := i.db.NamedExec(stmt, rr)
+	_, err := tx.NamedExec(stmt, rr)
 	if err != nil {
-		return types.Int{}, fmt.Errorf("failed to insert incident rule: %w", err)
+		return types.Int{}, err
 	}
 
-	return i.AddHistory(hr, true)
+	return i.AddHistory(tx, hr, true)
 }
 
-func (i *Incident) AddSourceSeverity(severity event.Severity, sourceID int64) error {
+func (i *Incident) AddSourceSeverity(tx *sqlx.Tx, severity event.Severity, sourceID int64) error {
 	i.SeverityBySource[sourceID] = severity
 
 	sourceSeverity := &SourceSeverity{
@@ -153,7 +151,7 @@ func (i *Incident) AddSourceSeverity(severity event.Severity, sourceID int64) er
 	}
 
 	stmt, _ := i.db.BuildUpsertStmt(sourceSeverity)
-	_, err := i.db.NamedExec(stmt, sourceSeverity)
+	_, err := tx.NamedExec(stmt, sourceSeverity)
 
 	return err
 }