gochat/internal/service/search_indexer_worker.go

package service

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"

	"github.com/gochat/gochat/internal/model"
	"github.com/gochat/gochat/internal/worker"
	"gorm.io/gorm"
)

const TaskTypeSearchIndex = "search:index"

type searchIndexJob struct {
	Operation string `json:"operation"`
	Entity    string `json:"entity"`
	AccountID uint   `json:"account_id"`
	ID        uint   `json:"id"`
}

// DurableSearchIndexer defers Meilisearch write-side synchronization to the
// durable worker while keeping read paths pointed at the real SearchService.
type DurableSearchIndexer struct {
	db       *gorm.DB
	worker   *worker.WorkerPool
	delegate SearchIndexer
}

func NewDurableSearchIndexer(db *gorm.DB, wp *worker.WorkerPool, delegate SearchIndexer) *DurableSearchIndexer {
	indexer := &DurableSearchIndexer{db: db, worker: wp, delegate: delegate}
	if wp != nil {
		wp.Register(TaskTypeSearchIndex, indexer.perform)
	}
	return indexer
}

func (i *DurableSearchIndexer) IndexConversation(ctx context.Context, conversation *model.Conversation) error {
	if conversation == nil {
		return nil
	}
	return i.enqueueOrIndex(ctx, "conversation", conversation.AccountID, conversation.ID, func() error {
		return i.delegate.IndexConversation(ctx, conversation)
	})
}

func (i *DurableSearchIndexer) DeleteConversation(ctx context.Context, accountID uint, id uint) error {
	return i.enqueueOrDelete(ctx, "conversation", accountID, id, func() error { return i.delegate.DeleteConversation(ctx, accountID, id) })
}

func (i *DurableSearchIndexer) IndexMessage(ctx context.Context, message *model.Message) error {
	if message == nil {
		return nil
	}
	return i.enqueueOrIndex(ctx, "message", message.AccountID, message.ID, func() error {
		return i.delegate.IndexMessage(ctx, message)
	})
}

func (i *DurableSearchIndexer) DeleteMessage(ctx context.Context, accountID uint, id uint) error {
	return i.enqueueOrDelete(ctx, "message", accountID, id, func() error { return i.delegate.DeleteMessage(ctx, accountID, id) })
}

func (i *DurableSearchIndexer) IndexContact(ctx context.Context, contact *model.Contact) error {
	if contact == nil {
		return nil
	}
	return i.enqueueOrIndex(ctx, "contact", contact.AccountID, contact.ID, func() error {
		return i.delegate.IndexContact(ctx, contact)
	})
}

func (i *DurableSearchIndexer) DeleteContact(ctx context.Context, accountID uint, id uint) error {
	return i.enqueueOrDelete(ctx, "contact", accountID, id, func() error { return i.delegate.DeleteContact(ctx, accountID, id) })
}

func (i *DurableSearchIndexer) IndexCompany(ctx context.Context, company *model.Company) error {
	if company == nil {
		return nil
	}
	return i.enqueueOrIndex(ctx, "company", company.AccountID, company.ID, func() error {
		return i.delegate.IndexCompany(ctx, company)
	})
}

func (i *DurableSearchIndexer) DeleteCompany(ctx context.Context, accountID uint, id uint) error {
	return i.enqueueOrDelete(ctx, "company", accountID, id, func() error { return i.delegate.DeleteCompany(ctx, accountID, id) })
}

func (i *DurableSearchIndexer) IndexArticle(ctx context.Context, article *model.Article) error {
	if article == nil {
		return nil
	}
	return i.enqueueOrIndex(ctx, "article", article.AccountID, article.ID, func() error {
		return i.delegate.IndexArticle(ctx, article)
	})
}

func (i *DurableSearchIndexer) DeleteArticle(ctx context.Context, accountID uint, id uint) error {
	return i.enqueueOrDelete(ctx, "article", accountID, id, func() error { return i.delegate.DeleteArticle(ctx, accountID, id) })
}

func (i *DurableSearchIndexer) enqueueOrIndex(ctx context.Context, entity string, accountID, id uint, fallback func() error) error {
	return i.enqueueOrRun(ctx, searchIndexJob{Operation: "index", Entity: entity, AccountID: accountID, ID: id}, fallback)
}

func (i *DurableSearchIndexer) enqueueOrDelete(ctx context.Context, entity string, accountID, id uint, fallback func() error) error {
	return i.enqueueOrRun(ctx, searchIndexJob{Operation: "delete", Entity: entity, AccountID: accountID, ID: id}, fallback)
}

func (i *DurableSearchIndexer) enqueueOrRun(ctx context.Context, payload searchIndexJob, fallback func() error) error {
	if i.delegate == nil {
		return nil
	}
	if i.worker == nil {
		return fallback()
	}
	_, err := i.worker.Enqueue(ctx, TaskTypeSearchIndex, payload, worker.WithQueue("search"), worker.WithMaxAttempts(3))
	return err
}

func (i *DurableSearchIndexer) perform(ctx context.Context, job *model.BackgroundJob) error {
	if i.delegate == nil {
		return nil
	}
	var payload searchIndexJob
	if err := json.Unmarshal(job.Payload, &payload); err != nil {
		return fmt.Errorf("unmarshal search index job: %w", err)
	}
	if payload.ID == 0 || payload.Entity == "" || payload.Operation == "" {
		return fmt.Errorf("invalid search index job payload: %#v", payload)
	}
	switch payload.Operation {
	case "index":
		return i.performIndex(ctx, payload)
	case "delete":
		return i.performDelete(ctx, payload)
	default:
		return fmt.Errorf("unsupported search index operation %q", payload.Operation)
	}
}

func (i *DurableSearchIndexer) performIndex(ctx context.Context, payload searchIndexJob) error {
	switch payload.Entity {
	case "conversation":
		var item model.Conversation
		if err := i.load(ctx, payload, &item); err != nil {
			return err
		}
		return i.delegate.IndexConversation(ctx, &item)
	case "message":
		var item model.Message
		if err := i.load(ctx, payload, &item); err != nil {
			return err
		}
		return i.delegate.IndexMessage(ctx, &item)
	case "contact":
		var item model.Contact
		if err := i.load(ctx, payload, &item); err != nil {
			return err
		}
		return i.delegate.IndexContact(ctx, &item)
	case "company":
		var item model.Company
		if err := i.load(ctx, payload, &item); err != nil {
			return err
		}
		return i.delegate.IndexCompany(ctx, &item)
	case "article":
		var item model.Article
		if err := i.load(ctx, payload, &item); err != nil {
			return err
		}
		return i.delegate.IndexArticle(ctx, &item)
	default:
		return fmt.Errorf("unsupported search index entity %q", payload.Entity)
	}
}

func (i *DurableSearchIndexer) load(ctx context.Context, payload searchIndexJob, dest any) error {
	err := i.db.WithContext(ctx).Where("id = ? AND account_id = ?", payload.ID, payload.AccountID).First(dest).Error
	if errors.Is(err, gorm.ErrRecordNotFound) {
		return i.performDelete(ctx, payload)
	}
	return err
}

func (i *DurableSearchIndexer) performDelete(ctx context.Context, payload searchIndexJob) error {
	switch payload.Entity {
	case "conversation":
		return i.delegate.DeleteConversation(ctx, payload.AccountID, payload.ID)
	case "message":
		return i.delegate.DeleteMessage(ctx, payload.AccountID, payload.ID)
	case "contact":
		return i.delegate.DeleteContact(ctx, payload.AccountID, payload.ID)
	case "company":
		return i.delegate.DeleteCompany(ctx, payload.AccountID, payload.ID)
	case "article":
		return i.delegate.DeleteArticle(ctx, payload.AccountID, payload.ID)
	default:
		return fmt.Errorf("unsupported search index entity %q", payload.Entity)
	}
}