Changeset 0.22.8 (#150)

server/extraction/queue.go

@@ -25,7 +25,7 @@ const (
 // QueueItem is the status.json written to the processing directory.
 // The sidecar extractor watches this directory for pending items.
 type QueueItem struct {
-	AttachmentID string `json:"attachment_id"`
+	FileID string `json:"file_id"`
 	StorageKey   string `json:"storage_key"`
 	ContentType  string `json:"content_type"`
 	Filename     string `json:"filename"`
@@ -40,7 +40,7 @@ type QueueItem struct {
 // ── Queue Manager ──────────────────────────
 
 // Queue manages the filesystem-based extraction queue.
-// Processing state lives in {storagePath}/processing/{attachment_id}/status.json.
+// Processing state lives in {storagePath}/processing/{file_id}/status.json.
 // The sidecar extractor watches for pending items and updates status.
 type Queue struct {
 	storagePath string
@@ -69,19 +69,19 @@ func NewQueue(storagePath string, concurrency int) (*Queue, error) {
 	}, nil
 }
 
-// Enqueue adds an attachment to the extraction queue.
+// Enqueue adds a file to the extraction queue.
 // Creates {storagePath}/processing/{id}/status.json with status "pending".
-func (q *Queue) Enqueue(attachmentID, storageKey, contentType, filename string) error {
+func (q *Queue) Enqueue(fileID, storageKey, contentType, filename string) error {
 	q.mu.Lock()
 	defer q.mu.Unlock()
 
-	itemDir := filepath.Join(q.storagePath, "processing", attachmentID)
+	itemDir := filepath.Join(q.storagePath, "processing", fileID)
 	if err := os.MkdirAll(itemDir, 0750); err != nil {
 		return fmt.Errorf("extraction: create item dir: %w", err)
 	}
 
 	item := QueueItem{
-		AttachmentID: attachmentID,
+		FileID: fileID,
 		StorageKey:   storageKey,
 		ContentType:  contentType,
 		Filename:     filename,
@@ -89,12 +89,12 @@ func (q *Queue) Enqueue(attachmentID, storageKey, contentType, filename string)
 		QueuedAt:     time.Now().UTC().Format(time.RFC3339),
 	}
 
-	return q.writeStatus(attachmentID, &item)
+	return q.writeStatus(fileID, &item)
 }
 
-// GetStatus reads the current extraction status for an attachment.
-func (q *Queue) GetStatus(attachmentID string) (*QueueItem, error) {
-	statusPath := filepath.Join(q.storagePath, "processing", attachmentID, "status.json")
+// GetStatus reads the current extraction status for a file.
+func (q *Queue) GetStatus(fileID string) (*QueueItem, error) {
+	statusPath := filepath.Join(q.storagePath, "processing", fileID, "status.json")
 	data, err := os.ReadFile(statusPath)
 	if err != nil {
 		if os.IsNotExist(err) {
@@ -111,40 +111,40 @@ func (q *Queue) GetStatus(attachmentID string) (*QueueItem, error) {
 }
 
 // MarkComplete updates status to complete and records the output key.
-func (q *Queue) MarkComplete(attachmentID, outputKey string) error {
+func (q *Queue) MarkComplete(fileID, outputKey string) error {
 	q.mu.Lock()
 	defer q.mu.Unlock()
 
-	item, err := q.GetStatus(attachmentID)
+	item, err := q.GetStatus(fileID)
 	if err != nil || item == nil {
-		return fmt.Errorf("extraction: item not found: %s", attachmentID)
+		return fmt.Errorf("extraction: item not found: %s", fileID)
 	}
 
 	item.Status = StatusComplete
 	item.OutputKey = outputKey
 	item.CompletedAt = time.Now().UTC().Format(time.RFC3339)
-	return q.writeStatus(attachmentID, item)
+	return q.writeStatus(fileID, item)
 }
 
 // MarkFailed updates status to failed with an error message.
-func (q *Queue) MarkFailed(attachmentID, errMsg string) error {
+func (q *Queue) MarkFailed(fileID, errMsg string) error {
 	q.mu.Lock()
 	defer q.mu.Unlock()
 
-	item, err := q.GetStatus(attachmentID)
+	item, err := q.GetStatus(fileID)
 	if err != nil || item == nil {
-		return fmt.Errorf("extraction: item not found: %s", attachmentID)
+		return fmt.Errorf("extraction: item not found: %s", fileID)
 	}
 
 	item.Status = StatusFailed
 	item.Error = errMsg
 	item.CompletedAt = time.Now().UTC().Format(time.RFC3339)
-	return q.writeStatus(attachmentID, item)
+	return q.writeStatus(fileID, item)
 }
 
 // Cleanup removes the processing directory for a completed/failed item.
-func (q *Queue) Cleanup(attachmentID string) error {
-	itemDir := filepath.Join(q.storagePath, "processing", attachmentID)
+func (q *Queue) Cleanup(fileID string) error {
+	itemDir := filepath.Join(q.storagePath, "processing", fileID)
 	return os.RemoveAll(itemDir)
 }
 
@@ -227,8 +227,8 @@ func (q *Queue) ListAll() ([]QueueItem, error) {
 
 // ── Internal Helpers ───────────────────────
 
-func (q *Queue) writeStatus(attachmentID string, item *QueueItem) error {
-	statusPath := filepath.Join(q.storagePath, "processing", attachmentID, "status.json")
+func (q *Queue) writeStatus(fileID string, item *QueueItem) error {
+	statusPath := filepath.Join(q.storagePath, "processing", fileID, "status.json")
 	data, err := json.MarshalIndent(item, "", "  ")
 	if err != nil {
 		return err