task 6 without facade [fix - 1]

Author: bobturanov

config.yml

@@ -49,8 +49,8 @@ kafka:
   brokers:
     - "localhost:9095"
     - "localhost:9096"
-    - "kafka-1:9092"
-    - "kafka-2:9092"
+#    - "kafka-1:9092"
+#    - "kafka-2:9092"
 
 telemetry:
   graylogPath: 127.0.0.1:12201

internal/app/repo/method_lock.go

@@ -33,18 +33,19 @@ func (r repo) Lock(ctx context.Context, n uint64) ([]model.VerificationEvent, er
 }
 
 func (r repo) convertToVerificationEventModel(n uint64, eventsData []struct {
-	EventId          uint64
-	EventType        model.EventType
-	EventStatus      model.EventStatus
-	VerificationId   uint64
-	VerificationName string
+	EventId          uint64 `db:"event_id"`
+	EventType        model.EventType `db:"type"`
+	EventStatus      model.EventStatus `db:"status"`
+	VerificationId   uint64 `db:"id"`
+	VerificationName string `db:"name"`
 }) []model.VerificationEvent {
 
 	events := make([]model.VerificationEvent, 0, n)
 
 	for _, event := range eventsData {
 		events = append(events, model.VerificationEvent{
 			ID:     event.EventId,
+			VerificationID: event.VerificationId,
 			Type:   event.EventType,
 			Status: event.EventStatus,
 			Entity: &model.Verification{
@@ -57,33 +58,34 @@ func (r repo) convertToVerificationEventModel(n uint64, eventsData []struct {
 }
 
 func (r repo) getEventsDataFromDB(ctx context.Context, err error, eventIds []uint64) ([]struct {
-	EventId          uint64
-	EventType        model.EventType
-	EventStatus      model.EventStatus
-	VerificationId   uint64
-	VerificationName string
+	EventId          uint64 `db:"event_id"`
+	EventType        model.EventType `db:"type"`
+	EventStatus      model.EventStatus `db:"status"`
+	VerificationId   uint64 `db:"id"`
+	VerificationName string `db:"name"`
 }, error) {
 
-	query, args, err := squirrel.Select("verification_events.id",
-		"verification_events.event_type",
-		"verification_events.event_status",
+	query, args, err := squirrel.Select("verification_events.event_id",
+		"verification_events.type",
+		"verification_events.status",
 		"verification.id",
 		"verification.name").
+		PlaceholderFormat(squirrel.Dollar).
 		Join("verification on verification.id = verification_events.verification_id").
 		From("verification_events").
-		Where(squirrel.Eq{"verification_events.id": eventIds}).
+		Where(squirrel.Eq{"verification_events.event_id": eventIds}).
 		ToSql()
 
 	if err != nil {
 		return nil, err
 	}
 
 	var eventsData []struct {
-		EventId          uint64
-		EventType        model.EventType
-		EventStatus      model.EventStatus
-		VerificationId   uint64
-		VerificationName string
+		EventId          uint64 `db:"event_id"`
+		EventType        model.EventType `db:"type"`
+		EventStatus      model.EventStatus `db:"status"`
+		VerificationId   uint64 `db:"id"`
+		VerificationName string `db:"name"`
 	}
 
 	err = r.db.SelectContext(ctx, &eventsData, query, args...)
@@ -96,22 +98,34 @@ func (r repo) getEventsDataFromDB(ctx context.Context, err error, eventIds []uin
 }
 
 func (r repo) getEventIdsFromDB(ctx context.Context, n uint64) ([]uint64, error) {
-	query, args, err := squirrel.Update("verification_events").
-		Set("event_status", model.Processed).
+	eventIds := make([]uint64, 0, n)
+	query, args, err := squirrel.Select("event_id").
+		From("verification_events").
 		PlaceholderFormat(squirrel.Dollar).
-		Where(squirrel.Eq{"event_status": model.Deferred}).
+		Where(squirrel.Eq{"status ": "DEFERRED"}).
 		Limit(n).
-		Suffix("RETURNING id").
 		ToSql()
 
 	if err != nil {
 		logger.ErrorKV(ctx, "repo.getEventIdsFromDB() get select query", "err", err)
 		return nil, err
 	}
 
-	eventIds := make([]uint64, 0, n)
 	err = r.db.SelectContext(ctx, &eventIds, query, args...)
 
+	query, args, err = squirrel.Update("verification_events").
+		Set("status", model.Processed).
+		PlaceholderFormat(squirrel.Dollar).
+		Where(squirrel.Eq{"event_id ": eventIds}).
+		ToSql()
+
+	if err != nil {
+		logger.ErrorKV(ctx, "repo.getEventIdsFromDB() get select query", "err", err)
+		return nil, err
+	}
+
+	_, err = r.db.ExecContext(ctx, query, args...)
+
 	if err != nil {
 		logger.ErrorKV(ctx, "repo.getEventIdsFromDB() get result query", "err", err)
 		return nil, err

internal/app/repo/method_remove.go

@@ -16,7 +16,7 @@ func (r repo) Remove(ctx context.Context, eventIDs []uint64) error {
 	query, args, err := squirrel.Update("verification_events").
 		PlaceholderFormat(squirrel.Dollar).
 		Set("status", model.Processed).
-		Where(squirrel.Eq{"id": eventIDs}).
+		Where(squirrel.Eq{"event_id": eventIDs}).
 		ToSql()
 
 	if err != nil {

internal/app/repo/method_unlock.go

@@ -17,7 +17,7 @@ func (r repo) Unlock(ctx context.Context, eventIDs []uint64) error {
 		PlaceholderFormat(squirrel.Dollar).
 		Set("status", model.Deferred).
 		Set("updated_at", "NOW()").
-		Where(squirrel.Eq{"id": eventIDs}).
+		Where(squirrel.Eq{"event_id": eventIDs}).
 		ToSql()
 
 	if err != nil {

migrations/00002_init_db.sql

@@ -3,7 +3,7 @@ CREATE TYPE event_type AS ENUM ('CREATED', 'UPDATED', 'REMOVED');
 CREATE TYPE event_status AS ENUM ('DEFERRED', 'PROCESSED');
 
 CREATE TABLE verification_events (
-                                id BIGSERIAL PRIMARY KEY,
+                                event_id BIGSERIAL PRIMARY KEY,
                                 verification_id BIGSERIAL NOT NULL,
                                 type event_type NOT NULL,
                                 status event_status NOT NULL,