455 lines
14 KiB
Go
455 lines
14 KiB
Go
package service
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/gochat/gochat/internal/llm"
|
|
"github.com/gochat/gochat/internal/model"
|
|
"github.com/gochat/gochat/internal/repository"
|
|
"github.com/gochat/gochat/internal/worker"
|
|
applogger "github.com/gochat/gochat/pkg/logger"
|
|
)
|
|
|
|
// CaptainDocumentService implements business logic for CaptainDocument operations.
|
|
// Reference: Chatwoot enterprise/app/controllers/api/v1/captain/documents_controller.rb
|
|
type CaptainDocumentService struct {
|
|
documentRepo *repository.CaptainDocumentRepo
|
|
assistantRepo *repository.CaptainAssistantRepo
|
|
llmProvider llm.Provider
|
|
syncBackend CaptainDocumentSyncBackend
|
|
worker *worker.WorkerPool
|
|
}
|
|
|
|
type CaptainDocumentSyncBackend interface {
|
|
SyncCaptainDocument(ctx context.Context, doc *model.CaptainDocument) (*CaptainDocumentSyncResult, error)
|
|
}
|
|
|
|
type CaptainDocumentSyncResult struct {
|
|
Content string
|
|
Title string
|
|
ErrorCode string
|
|
}
|
|
|
|
// NewCaptainDocumentService creates a new CaptainDocumentService.
|
|
func NewCaptainDocumentService(
|
|
documentRepo *repository.CaptainDocumentRepo,
|
|
llmProvider llm.Provider,
|
|
assistantRepo ...*repository.CaptainAssistantRepo,
|
|
) *CaptainDocumentService {
|
|
s := &CaptainDocumentService{
|
|
documentRepo: documentRepo,
|
|
llmProvider: llmProvider,
|
|
}
|
|
if len(assistantRepo) > 0 {
|
|
s.assistantRepo = assistantRepo[0]
|
|
}
|
|
return s
|
|
}
|
|
|
|
func (s *CaptainDocumentService) SetSyncBackend(syncBackend CaptainDocumentSyncBackend) {
|
|
s.syncBackend = syncBackend
|
|
}
|
|
|
|
func (s *CaptainDocumentService) SetWorkerPool(wp *worker.WorkerPool) {
|
|
s.worker = wp
|
|
RegisterCaptainDocumentJobs(wp, s)
|
|
}
|
|
|
|
// --- Request DTOs ---
|
|
|
|
// CreateDocumentRequest is the DTO for creating a document.
|
|
type CreateDocumentRequest struct {
|
|
Name string `json:"name" validate:"required"`
|
|
ExternalLink string `json:"external_link" validate:"required"`
|
|
AssistantID uint `json:"assistant_id"`
|
|
}
|
|
|
|
// UpdateDocumentRequest is the DTO for updating a document.
|
|
type UpdateDocumentRequest struct {
|
|
Name string `json:"name"`
|
|
ExternalLink string `json:"external_link"`
|
|
}
|
|
|
|
type ListDocumentsRequest struct {
|
|
AssistantID uint
|
|
Page int
|
|
PerPage int
|
|
Filter string
|
|
Source string
|
|
Sort string
|
|
SearchKey string
|
|
}
|
|
|
|
// --- CRUD Operations ---
|
|
|
|
// Create creates a new CaptainDocument.
|
|
func (s *CaptainDocumentService) Create(ctx context.Context, assistantID, accountID uint, req *CreateDocumentRequest) (*model.CaptainDocument, error) {
|
|
if s.assistantRepo != nil {
|
|
if _, err := s.assistantRepo.GetByAccountAndID(ctx, accountID, assistantID); err != nil {
|
|
return nil, fmt.Errorf("assistant not found: %w", err)
|
|
}
|
|
}
|
|
doc := &model.CaptainDocument{
|
|
AccountID: accountID,
|
|
AssistantID: assistantID,
|
|
Name: req.Name,
|
|
ExternalLink: req.ExternalLink,
|
|
Status: model.DocumentStatusPending,
|
|
}
|
|
|
|
if err := s.documentRepo.Create(ctx, doc); err != nil {
|
|
applogger.L().Errorf("Create captain document: %v", err)
|
|
return nil, fmt.Errorf("create document: %w", err)
|
|
}
|
|
if created, err := s.documentRepo.GetByAccountAndID(ctx, accountID, doc.ID); err == nil {
|
|
return created, nil
|
|
}
|
|
return doc, nil
|
|
}
|
|
|
|
// Get retrieves a document by ID.
|
|
func (s *CaptainDocumentService) Get(ctx context.Context, id uint) (*model.CaptainDocument, error) {
|
|
doc, err := s.documentRepo.GetByID(ctx, id)
|
|
if err != nil {
|
|
applogger.L().Errorf("Get captain document: %v", err)
|
|
return nil, fmt.Errorf("get document: %w", err)
|
|
}
|
|
return doc, nil
|
|
}
|
|
|
|
func (s *CaptainDocumentService) GetByAccount(ctx context.Context, accountID, id uint) (*model.CaptainDocument, error) {
|
|
doc, err := s.documentRepo.GetByAccountAndID(ctx, accountID, id)
|
|
if err != nil {
|
|
applogger.L().Errorf("Get captain document: %v", err)
|
|
return nil, fmt.Errorf("get document: %w", err)
|
|
}
|
|
return doc, nil
|
|
}
|
|
|
|
// Update updates an existing document.
|
|
func (s *CaptainDocumentService) Update(ctx context.Context, id uint, req *UpdateDocumentRequest) (*model.CaptainDocument, error) {
|
|
doc, err := s.documentRepo.GetByID(ctx, id)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("document not found: %w", err)
|
|
}
|
|
|
|
if req.Name != "" {
|
|
doc.Name = req.Name
|
|
}
|
|
if req.ExternalLink != "" {
|
|
doc.ExternalLink = req.ExternalLink
|
|
}
|
|
|
|
if err := s.documentRepo.Update(ctx, doc); err != nil {
|
|
applogger.L().Errorf("Update captain document: %v", err)
|
|
return nil, fmt.Errorf("update document: %w", err)
|
|
}
|
|
return doc, nil
|
|
}
|
|
|
|
// Delete deletes a document by ID.
|
|
func (s *CaptainDocumentService) Delete(ctx context.Context, id uint) error {
|
|
if err := s.documentRepo.Delete(ctx, id); err != nil {
|
|
applogger.L().Errorf("Delete captain document: %v", err)
|
|
return fmt.Errorf("delete document: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *CaptainDocumentService) DeleteByAccount(ctx context.Context, accountID, id uint) error {
|
|
if _, err := s.documentRepo.GetByAccountAndID(ctx, accountID, id); err != nil {
|
|
return fmt.Errorf("document not found: %w", err)
|
|
}
|
|
if err := s.documentRepo.DeleteByAccount(ctx, accountID, id); err != nil {
|
|
applogger.L().Errorf("Delete captain document: %v", err)
|
|
return fmt.Errorf("delete document: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// List retrieves documents for an assistant with pagination.
|
|
func (s *CaptainDocumentService) List(ctx context.Context, assistantID uint, offset, limit int) ([]model.CaptainDocument, int64, error) {
|
|
docs, count, err := s.documentRepo.ListByAssistant(ctx, assistantID, offset, limit)
|
|
if err != nil {
|
|
applogger.L().Errorf("List captain documents: %v", err)
|
|
return nil, 0, fmt.Errorf("list documents: %w", err)
|
|
}
|
|
return docs, count, nil
|
|
}
|
|
|
|
func (s *CaptainDocumentService) ListByAccount(ctx context.Context, accountID uint, req ListDocumentsRequest) ([]model.CaptainDocument, int64, int, error) {
|
|
page := req.Page
|
|
if page <= 0 {
|
|
page = 1
|
|
}
|
|
perPage := req.PerPage
|
|
if perPage <= 0 {
|
|
perPage = 25
|
|
}
|
|
docs, count, err := s.documentRepo.ListByAccount(ctx, accountID, repository.CaptainDocumentListFilters{
|
|
AssistantID: req.AssistantID,
|
|
Source: req.Source,
|
|
Filter: req.Filter,
|
|
SearchKey: req.SearchKey,
|
|
Sort: req.Sort,
|
|
Offset: (page - 1) * perPage,
|
|
Limit: perPage,
|
|
})
|
|
if err != nil {
|
|
applogger.L().Errorf("List captain documents: %v", err)
|
|
return nil, 0, page, fmt.Errorf("list documents: %w", err)
|
|
}
|
|
return docs, count, page, nil
|
|
}
|
|
|
|
func (s *CaptainDocumentService) MarkSyncing(ctx context.Context, accountID, id uint) (*model.CaptainDocument, error) {
|
|
doc, err := s.documentRepo.GetByAccountAndID(ctx, accountID, id)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("document not found: %w", err)
|
|
}
|
|
now := time.Now().Unix()
|
|
doc.LastSyncAttemptedAt = &now
|
|
doc.SyncStatus = model.DocumentSyncStatusPending
|
|
doc.LastSyncErrorCode = ""
|
|
if err := s.documentRepo.Update(ctx, doc); err != nil {
|
|
return nil, fmt.Errorf("mark document syncing: %w", err)
|
|
}
|
|
return s.documentRepo.GetByAccountAndID(ctx, accountID, id)
|
|
}
|
|
|
|
func (s *CaptainDocumentService) RequestSyncDocumentByAccount(ctx context.Context, accountID, id uint) (*model.CaptainDocument, error) {
|
|
doc, err := s.MarkSyncing(ctx, accountID, id)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if s.worker == nil {
|
|
return doc, nil
|
|
}
|
|
if _, err := s.worker.Enqueue(ctx, TaskTypeCaptainDocumentSync, captainDocumentSyncJob{AccountID: accountID, DocumentID: id}, worker.WithQueue("low"), worker.WithMaxAttempts(3)); err != nil {
|
|
return nil, err
|
|
}
|
|
return doc, nil
|
|
}
|
|
|
|
func (s *CaptainDocumentService) SyncDocumentByAccount(ctx context.Context, accountID, id uint) (*model.CaptainDocument, error) {
|
|
doc, err := s.documentRepo.GetByAccountAndID(ctx, accountID, id)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("document not found: %w", err)
|
|
}
|
|
if err := s.markDocumentSyncStarted(ctx, doc); err != nil {
|
|
return nil, err
|
|
}
|
|
if s.syncBackend == nil {
|
|
return s.markDocumentSyncFailed(ctx, accountID, id, "sync_disabled")
|
|
}
|
|
|
|
result, err := s.syncBackend.SyncCaptainDocument(ctx, doc)
|
|
if err != nil {
|
|
updated, markErr := s.markDocumentSyncFailed(ctx, accountID, id, "sync_error")
|
|
if markErr != nil {
|
|
return nil, markErr
|
|
}
|
|
return updated, fmt.Errorf("sync document: %w", err)
|
|
}
|
|
if result == nil {
|
|
return s.markDocumentSyncFailed(ctx, accountID, id, "sync_error")
|
|
}
|
|
if result.ErrorCode != "" {
|
|
return s.markDocumentSyncFailed(ctx, accountID, id, result.ErrorCode)
|
|
}
|
|
if strings.TrimSpace(result.Content) == "" {
|
|
return s.markDocumentSyncFailed(ctx, accountID, id, "content_empty")
|
|
}
|
|
|
|
doc.Content = strings.TrimSpace(result.Content)
|
|
if strings.TrimSpace(result.Title) != "" {
|
|
doc.Name = strings.TrimSpace(result.Title)
|
|
}
|
|
doc.ContentFingerprint = computeFingerprint(doc.Content)
|
|
doc.Status = model.DocumentStatusCompleted
|
|
doc.SyncStatus = model.DocumentSyncStatusSynced
|
|
doc.LastSyncErrorCode = ""
|
|
now := time.Now().Unix()
|
|
doc.LastSyncedAt = &now
|
|
doc.LastSyncAttemptedAt = &now
|
|
if err := s.documentRepo.Update(ctx, doc); err != nil {
|
|
return nil, fmt.Errorf("update synced document: %w", err)
|
|
}
|
|
return s.documentRepo.GetByAccountAndID(ctx, accountID, id)
|
|
}
|
|
|
|
func (s *CaptainDocumentService) markDocumentSyncStarted(ctx context.Context, doc *model.CaptainDocument) error {
|
|
now := time.Now().Unix()
|
|
doc.LastSyncAttemptedAt = &now
|
|
doc.SyncStatus = model.DocumentSyncStatusPending
|
|
doc.LastSyncErrorCode = ""
|
|
if err := s.documentRepo.Update(ctx, doc); err != nil {
|
|
return fmt.Errorf("mark document syncing: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *CaptainDocumentService) markDocumentSyncFailed(ctx context.Context, accountID, id uint, code string) (*model.CaptainDocument, error) {
|
|
doc, err := s.documentRepo.GetByAccountAndID(ctx, accountID, id)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("document not found: %w", err)
|
|
}
|
|
now := time.Now().Unix()
|
|
doc.SyncStatus = model.DocumentSyncStatusFailed
|
|
doc.LastSyncAttemptedAt = &now
|
|
doc.LastSyncErrorCode = code
|
|
if err := s.documentRepo.Update(ctx, doc); err != nil {
|
|
return nil, fmt.Errorf("mark document sync failed: %w", err)
|
|
}
|
|
return s.documentRepo.GetByAccountAndID(ctx, accountID, id)
|
|
}
|
|
|
|
// --- Document Processing ---
|
|
|
|
// ProcessDocument extracts content and generates embedding for a document.
|
|
// Reference: Chatwoot Captain::DocumentProcessorJob
|
|
func (s *CaptainDocumentService) ProcessDocument(ctx context.Context, id uint) error {
|
|
doc, err := s.documentRepo.GetByID(ctx, id)
|
|
if err != nil {
|
|
return fmt.Errorf("document not found: %w", err)
|
|
}
|
|
|
|
// Mark status as in_progress
|
|
doc.Status = model.DocumentStatusInProgress
|
|
if err := s.documentRepo.Update(ctx, doc); err != nil {
|
|
return fmt.Errorf("update document status: %w", err)
|
|
}
|
|
|
|
// Fetch content from external link
|
|
content, err := s.fetchContent(ctx, doc.ExternalLink)
|
|
if err != nil {
|
|
doc.Status = model.DocumentStatusFailed
|
|
doc.LastSyncErrorCode = "fetch_failed"
|
|
s.documentRepo.Update(ctx, doc)
|
|
applogger.L().Errorf("ProcessDocument fetch content: %v", err)
|
|
return fmt.Errorf("fetch content: %w", err)
|
|
}
|
|
|
|
// Compute content fingerprint for deduplication
|
|
fingerprint := computeFingerprint(content)
|
|
|
|
// Update document with content
|
|
doc.Content = content
|
|
doc.ContentFingerprint = fingerprint
|
|
doc.Status = model.DocumentStatusCompleted
|
|
now := time.Now().Unix()
|
|
doc.LastSyncedAt = &now
|
|
doc.SyncStatus = model.DocumentSyncStatusSynced
|
|
|
|
if err := s.documentRepo.Update(ctx, doc); err != nil {
|
|
applogger.L().Errorf("ProcessDocument update: %v", err)
|
|
return fmt.Errorf("update document: %w", err)
|
|
}
|
|
|
|
// Generate embedding for the document content
|
|
_, err = s.llmProvider.CreateEmbedding(ctx, llm.EmbeddingRequest{
|
|
Input: []string{content},
|
|
Model: "text-embedding-ada-002",
|
|
})
|
|
if err != nil {
|
|
applogger.L().Errorf("ProcessDocument embedding: %v", err)
|
|
// Embedding failure does not block document processing
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// SyncDocument re-fetches content from the external URL and updates the document.
|
|
// Reference: Chatwoot Captain::DocumentSyncJob
|
|
func (s *CaptainDocumentService) SyncDocument(ctx context.Context, id uint) error {
|
|
doc, err := s.documentRepo.GetByID(ctx, id)
|
|
if err != nil {
|
|
return fmt.Errorf("document not found: %w", err)
|
|
}
|
|
|
|
nowAttempt := time.Now().Unix()
|
|
doc.LastSyncAttemptedAt = &nowAttempt
|
|
doc.SyncStatus = model.DocumentSyncStatusPending
|
|
s.documentRepo.Update(ctx, doc)
|
|
|
|
content, err := s.fetchContent(ctx, doc.ExternalLink)
|
|
if err != nil {
|
|
doc.SyncStatus = model.DocumentSyncStatusFailed
|
|
doc.LastSyncErrorCode = "fetch_failed"
|
|
s.documentRepo.Update(ctx, doc)
|
|
applogger.L().Errorf("SyncDocument fetch: %v", err)
|
|
return fmt.Errorf("sync content: %w", err)
|
|
}
|
|
|
|
fingerprint := computeFingerprint(content)
|
|
|
|
// If content unchanged, mark as synced (no update needed)
|
|
if fingerprint == doc.ContentFingerprint {
|
|
now := time.Now().Unix()
|
|
doc.LastSyncedAt = &now
|
|
doc.SyncStatus = model.DocumentSyncStatusSynced
|
|
s.documentRepo.Update(ctx, doc)
|
|
return nil
|
|
}
|
|
|
|
// Content changed, update document
|
|
doc.Content = content
|
|
doc.ContentFingerprint = fingerprint
|
|
now := time.Now().Unix()
|
|
doc.LastSyncedAt = &now
|
|
doc.SyncStatus = model.DocumentSyncStatusSynced
|
|
doc.LastSyncErrorCode = ""
|
|
|
|
if err := s.documentRepo.Update(ctx, doc); err != nil {
|
|
applogger.L().Errorf("SyncDocument update: %v", err)
|
|
return fmt.Errorf("update synced document: %w", err)
|
|
}
|
|
|
|
// Re-generate embedding for updated content
|
|
_, err = s.llmProvider.CreateEmbedding(ctx, llm.EmbeddingRequest{
|
|
Input: []string{content},
|
|
Model: "text-embedding-ada-002",
|
|
})
|
|
if err != nil {
|
|
applogger.L().Errorf("SyncDocument embedding: %v", err)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// fetchContent retrieves content from an external URL.
|
|
func (s *CaptainDocumentService) fetchContent(ctx context.Context, url string) (string, error) {
|
|
client := &http.Client{Timeout: 30 * time.Second}
|
|
resp, err := client.Get(url)
|
|
if err != nil {
|
|
return "", fmt.Errorf("HTTP GET %s: %w", url, err)
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
if resp.StatusCode != http.StatusOK {
|
|
return "", fmt.Errorf("HTTP GET %s returned status %d", url, resp.StatusCode)
|
|
}
|
|
|
|
// Limit response body to 1MB
|
|
body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
|
|
if err != nil {
|
|
return "", fmt.Errorf("read response body: %w", err)
|
|
}
|
|
|
|
return strings.TrimSpace(string(body)), nil
|
|
}
|
|
|
|
// computeFingerprint generates a SHA256 hash of content for deduplication.
|
|
func computeFingerprint(content string) string {
|
|
h := sha256.New()
|
|
normalized := strings.Join(strings.Fields(content), " ")
|
|
h.Write([]byte(normalized))
|
|
return fmt.Sprintf("%x", h.Sum(nil))
|
|
}
|