Initial commit - Lost and Found Revisi
This commit is contained in:
bambang-code1
@@ -0,0 +1,78 @@
|
||||
// internal/workers/audit_worker.go - FIXED
|
||||
package workers
|
||||
|
||||
import (
|
||||
"log"
|
||||
"lost-and-found/internal/repositories"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
// AuditWorker handles audit log background tasks
|
||||
type AuditWorker struct {
|
||||
db *gorm.DB
|
||||
auditLogRepo *repositories.AuditLogRepository
|
||||
stopChan chan bool
|
||||
}
|
||||
|
||||
// NewAuditWorker creates a new audit worker
|
||||
func NewAuditWorker(db *gorm.DB) *AuditWorker {
|
||||
return &AuditWorker{
|
||||
db: db,
|
||||
auditLogRepo: repositories.NewAuditLogRepository(db),
|
||||
stopChan: make(chan bool),
|
||||
}
|
||||
}
|
||||
|
||||
// Start starts the audit worker
|
||||
// ✅ FIXED: Now runs in goroutine (non-blocking)
|
||||
func (w *AuditWorker) Start() {
|
||||
log.Println("🔍 Audit Worker started")
|
||||
|
||||
// ✅ Run in goroutine to avoid blocking
|
||||
go func() {
|
||||
ticker := time.NewTicker(24 * time.Hour) // Run daily
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
w.cleanupOldLogs()
|
||||
case <-w.stopChan:
|
||||
log.Println("🔍 Audit Worker stopped")
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
// ✅ Return immediately after starting goroutine
|
||||
}
|
||||
|
||||
// Stop stops the audit worker
|
||||
func (w *AuditWorker) Stop() {
|
||||
close(w.stopChan)
|
||||
}
|
||||
|
||||
// cleanupOldLogs removes audit logs older than 1 year
|
||||
func (w *AuditWorker) cleanupOldLogs() {
|
||||
log.Println("🧹 Cleaning up old audit logs...")
|
||||
|
||||
cutoffDate := time.Now().AddDate(-1, 0, 0) // 1 year ago
|
||||
|
||||
result := w.db.Unscoped().Where("created_at < ?", cutoffDate).Delete(&struct {
|
||||
tableName struct{} `gorm:"audit_logs"`
|
||||
}{})
|
||||
|
||||
if result.Error != nil {
|
||||
log.Printf("❌ Failed to cleanup audit logs: %v", result.Error)
|
||||
return
|
||||
}
|
||||
|
||||
log.Printf("✅ Cleaned up %d old audit log entries", result.RowsAffected)
|
||||
}
|
||||
|
||||
// RunNow runs cleanup immediately (for testing)
|
||||
func (w *AuditWorker) RunNow() {
|
||||
log.Println("▶️ Running audit cleanup manually...")
|
||||
w.cleanupOldLogs()
|
||||
}
|
||||
@@ -0,0 +1,223 @@
|
||||
// internal/workers/expire_worker.go - IMPROVED VERSION
|
||||
package workers
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"lost-and-found/internal/models"
|
||||
"lost-and-found/internal/repositories"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
"gorm.io/gorm/clause"
|
||||
)
|
||||
|
||||
type ExpireWorker struct {
|
||||
db *gorm.DB
|
||||
itemRepo *repositories.ItemRepository
|
||||
archiveRepo *repositories.ArchiveRepository
|
||||
stopChan chan bool
|
||||
wg sync.WaitGroup
|
||||
mu sync.Mutex
|
||||
|
||||
// ✅ Worker pool configuration
|
||||
maxWorkers int
|
||||
taskQueue chan *models.Item
|
||||
}
|
||||
|
||||
func NewExpireWorker(db *gorm.DB) *ExpireWorker {
|
||||
return &ExpireWorker{
|
||||
db: db,
|
||||
itemRepo: repositories.NewItemRepository(db),
|
||||
archiveRepo: repositories.NewArchiveRepository(db),
|
||||
stopChan: make(chan bool),
|
||||
maxWorkers: 5, // ✅ Configurable worker count
|
||||
taskQueue: make(chan *models.Item, 100), // ✅ Buffered channel
|
||||
}
|
||||
}
|
||||
|
||||
// ✅ Start with proper worker pool
|
||||
func (w *ExpireWorker) Start() {
|
||||
w.wg.Add(1)
|
||||
|
||||
go func() {
|
||||
defer w.wg.Done()
|
||||
|
||||
log.Println("⏰ Expire Worker started with", w.maxWorkers, "workers")
|
||||
|
||||
// ✅ Start worker pool
|
||||
w.startWorkerPool()
|
||||
|
||||
// Run immediately on start
|
||||
w.expireItems()
|
||||
|
||||
// Then run every hour
|
||||
ticker := time.NewTicker(1 * time.Hour)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
w.expireItems()
|
||||
case <-w.stopChan:
|
||||
log.Println("⏰ Stopping Expire Worker...")
|
||||
close(w.taskQueue) // ✅ Close task queue to signal workers
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// ✅ Worker Pool Implementation
|
||||
func (w *ExpireWorker) startWorkerPool() {
|
||||
for i := 0; i < w.maxWorkers; i++ {
|
||||
w.wg.Add(1)
|
||||
go w.worker(i)
|
||||
}
|
||||
}
|
||||
|
||||
// ✅ Individual Worker
|
||||
func (w *ExpireWorker) worker(id int) {
|
||||
defer w.wg.Done()
|
||||
|
||||
log.Printf("🔧 Worker %d started", id)
|
||||
|
||||
for item := range w.taskQueue {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
|
||||
success := w.archiveExpiredItem(ctx, item)
|
||||
|
||||
if success {
|
||||
log.Printf("✅ Worker %d: Archived item ID %d (%s)", id, item.ID, item.Name)
|
||||
} else {
|
||||
log.Printf("❌ Worker %d: Failed to archive item ID %d", id, item.ID)
|
||||
}
|
||||
|
||||
cancel()
|
||||
}
|
||||
|
||||
log.Printf("🔧 Worker %d stopped", id)
|
||||
}
|
||||
|
||||
// ✅ Improved expireItems
|
||||
func (w *ExpireWorker) expireItems() {
|
||||
log.Println("🔍 Checking for expired items (Using Stored Procedure)...")
|
||||
|
||||
// Panggil method repository yang mengeksekusi SP
|
||||
// Pastikan Anda sudah menambahkan method CallArchiveExpiredProcedure di ItemRepository
|
||||
// seperti pada langkah 1 jawaban sebelumnya.
|
||||
count, err := w.itemRepo.CallArchiveExpiredProcedure()
|
||||
|
||||
if err != nil {
|
||||
log.Printf("❌ Error executing archive SP: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
if count > 0 {
|
||||
log.Printf("✅ Successfully archived %d items via DB Procedure", count)
|
||||
} else {
|
||||
log.Println("✅ No expired items found to archive")
|
||||
}
|
||||
}
|
||||
|
||||
func (w *ExpireWorker) findExpiredItems(ctx context.Context) ([]models.Item, error) {
|
||||
var items []models.Item
|
||||
err := w.db.WithContext(ctx).
|
||||
Where("expires_at <= ? AND status = ? AND deleted_at IS NULL",
|
||||
time.Now(), models.ItemStatusUnclaimed).
|
||||
Preload("Category").
|
||||
Find(&items).Error
|
||||
|
||||
return items, err
|
||||
}
|
||||
|
||||
// ✅ Archive with proper transaction & locking
|
||||
func (w *ExpireWorker) archiveExpiredItem(ctx context.Context, item *models.Item) bool {
|
||||
err := w.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||||
// ✅ Lock the item
|
||||
var lockedItem models.Item
|
||||
// Locking/Isolation: Menerapkan Pessimistic Lock (FOR UPDATE) pada baris item yang akan diproses, mencegah worker lain atau API call lain memodifikasi item ini selama transaksi berlangsung, menjaga Konsistensi dan Pemulihan.
|
||||
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).
|
||||
Where("id = ? AND deleted_at IS NULL", item.ID).
|
||||
First(&lockedItem).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Double check it's still unclaimed
|
||||
if lockedItem.Status != models.ItemStatusUnclaimed {
|
||||
return nil // Already processed
|
||||
}
|
||||
|
||||
// 1. Create archive record
|
||||
archive := &models.Archive{
|
||||
ItemID: item.ID,
|
||||
Name: item.Name,
|
||||
CategoryID: item.CategoryID,
|
||||
PhotoURL: item.PhotoURL,
|
||||
Location: item.Location,
|
||||
Description: item.Description,
|
||||
DateFound: item.DateFound,
|
||||
Status: models.ItemStatusExpired,
|
||||
ReporterName: item.ReporterName,
|
||||
ReporterContact: item.ReporterContact,
|
||||
ArchivedReason: models.ArchiveReasonExpired,
|
||||
ClaimedBy: nil,
|
||||
ArchivedAt: time.Now(),
|
||||
}
|
||||
|
||||
if err := tx.Create(archive).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// 2. Update item status to expired
|
||||
if err := tx.Model(&lockedItem).Update("status", models.ItemStatusExpired).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// 3. Create audit log
|
||||
auditLog := &models.AuditLog{
|
||||
UserID: nil,
|
||||
Action: "expire",
|
||||
EntityType: models.EntityItem,
|
||||
EntityID: &item.ID,
|
||||
Details: "Item automatically expired and archived by system",
|
||||
IPAddress: "system",
|
||||
UserAgent: "expire_worker",
|
||||
}
|
||||
if err := tx.Create(auditLog).Error; err != nil {
|
||||
log.Printf("Warning: failed to create audit log for item %d: %v", item.ID, err)
|
||||
}
|
||||
|
||||
return nil
|
||||
})
|
||||
|
||||
return err == nil
|
||||
}
|
||||
|
||||
// ✅ Graceful Stop with WaitGroup
|
||||
func (w *ExpireWorker) Stop() {
|
||||
log.Println("🛑 Signaling Expire Worker to stop...")
|
||||
w.stopChan <- true
|
||||
|
||||
log.Println("⏳ Waiting for all workers to finish...")
|
||||
w.wg.Wait()
|
||||
|
||||
log.Println("✅ Expire Worker gracefully stopped")
|
||||
}
|
||||
|
||||
// RunNow for manual trigger
|
||||
func (w *ExpireWorker) RunNow() {
|
||||
log.Println("▶️ Running expiration check manually...")
|
||||
|
||||
var wg sync.WaitGroup
|
||||
wg.Add(1)
|
||||
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
w.expireItems()
|
||||
}()
|
||||
|
||||
wg.Wait()
|
||||
log.Println("✅ Manual expiration check completed")
|
||||
}
|
||||
@@ -0,0 +1,95 @@
|
||||
// internal/workers/matching_worker.go - FIXED
|
||||
package workers
|
||||
|
||||
import (
|
||||
"log"
|
||||
"lost-and-found/internal/repositories"
|
||||
"lost-and-found/internal/services"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
// MatchingWorker handles automatic matching of lost and found items
|
||||
type MatchingWorker struct {
|
||||
db *gorm.DB
|
||||
matchService *services.MatchService
|
||||
itemRepo *repositories.ItemRepository
|
||||
lostItemRepo *repositories.LostItemRepository
|
||||
stopChan chan bool
|
||||
}
|
||||
|
||||
// NewMatchingWorker creates a new matching worker
|
||||
func NewMatchingWorker(db *gorm.DB) *MatchingWorker {
|
||||
return &MatchingWorker{
|
||||
db: db,
|
||||
matchService: services.NewMatchService(db),
|
||||
itemRepo: repositories.NewItemRepository(db),
|
||||
lostItemRepo: repositories.NewLostItemRepository(db),
|
||||
stopChan: make(chan bool),
|
||||
}
|
||||
}
|
||||
|
||||
// Start starts the matching worker
|
||||
// ✅ FIXED: Now runs in goroutine (non-blocking)
|
||||
func (w *MatchingWorker) Start() {
|
||||
log.Println("🔗 Matching Worker started")
|
||||
|
||||
// ✅ Run in goroutine to avoid blocking
|
||||
go func() {
|
||||
// Run every 30 minutes
|
||||
ticker := time.NewTicker(30 * time.Minute)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
w.performMatching()
|
||||
case <-w.stopChan:
|
||||
log.Println("🔗 Matching Worker stopped")
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
// ✅ Return immediately after starting goroutine
|
||||
}
|
||||
|
||||
// Stop stops the matching worker
|
||||
func (w *MatchingWorker) Stop() {
|
||||
close(w.stopChan)
|
||||
}
|
||||
|
||||
// performMatching performs automatic matching between lost and found items
|
||||
func (w *MatchingWorker) performMatching() {
|
||||
log.Println("🔍 Performing automatic matching...")
|
||||
|
||||
// Get all unclaimed items
|
||||
items, _, err := w.itemRepo.FindAll(1, 1000, "unclaimed", "", "")
|
||||
if err != nil {
|
||||
log.Printf("❌ Error fetching items: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
if len(items) == 0 {
|
||||
log.Println("✅ No unclaimed items to match")
|
||||
return
|
||||
}
|
||||
|
||||
matchCount := 0
|
||||
for _, item := range items {
|
||||
// Auto-match with lost items
|
||||
if err := w.matchService.AutoMatchNewItem(item.ID); err != nil {
|
||||
log.Printf("❌ Failed to match item ID %d: %v", item.ID, err)
|
||||
continue
|
||||
}
|
||||
matchCount++
|
||||
}
|
||||
|
||||
log.Printf("✅ Completed matching for %d items", matchCount)
|
||||
}
|
||||
|
||||
// RunNow runs matching immediately (for testing)
|
||||
func (w *MatchingWorker) RunNow() {
|
||||
log.Println("▶️ Running matching manually...")
|
||||
w.performMatching()
|
||||
}
|
||||
@@ -0,0 +1,122 @@
|
||||
// internal/workers/notification_worker.go - FIXED
|
||||
package workers
|
||||
|
||||
import (
|
||||
"log"
|
||||
"lost-and-found/internal/models"
|
||||
"lost-and-found/internal/repositories"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
// NotificationWorker handles sending notifications asynchronously
|
||||
type NotificationWorker struct {
|
||||
db *gorm.DB
|
||||
notificationRepo *repositories.NotificationRepository
|
||||
matchRepo *repositories.MatchResultRepository
|
||||
stopChan chan bool
|
||||
}
|
||||
|
||||
// NewNotificationWorker creates a new notification worker
|
||||
func NewNotificationWorker(db *gorm.DB) *NotificationWorker {
|
||||
return &NotificationWorker{
|
||||
db: db,
|
||||
notificationRepo: repositories.NewNotificationRepository(db),
|
||||
matchRepo: repositories.NewMatchResultRepository(db),
|
||||
stopChan: make(chan bool),
|
||||
}
|
||||
}
|
||||
|
||||
// Start starts the notification worker
|
||||
// ✅ FIXED: Now runs in goroutine (non-blocking)
|
||||
func (w *NotificationWorker) Start() {
|
||||
log.Println("📬 Notification Worker started")
|
||||
|
||||
// ✅ Run in goroutine to avoid blocking
|
||||
go func() {
|
||||
// Run every 5 minutes
|
||||
ticker := time.NewTicker(5 * time.Minute)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
w.sendPendingNotifications()
|
||||
case <-w.stopChan:
|
||||
log.Println("📬 Notification Worker stopped")
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
// ✅ Return immediately after starting goroutine
|
||||
}
|
||||
|
||||
// Stop stops the notification worker
|
||||
func (w *NotificationWorker) Stop() {
|
||||
close(w.stopChan)
|
||||
}
|
||||
|
||||
// sendPendingNotifications sends notifications for unnotified matches
|
||||
func (w *NotificationWorker) sendPendingNotifications() {
|
||||
log.Println("📧 Checking for pending notifications...")
|
||||
|
||||
// Get unnotified matches
|
||||
matches, err := w.matchRepo.FindUnnotifiedMatches()
|
||||
if err != nil {
|
||||
log.Printf("❌ Error fetching unnotified matches: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
if len(matches) == 0 {
|
||||
log.Println("✅ No pending notifications")
|
||||
return
|
||||
}
|
||||
|
||||
log.Printf("📧 Found %d pending notifications", len(matches))
|
||||
|
||||
sentCount := 0
|
||||
for _, match := range matches {
|
||||
// Send notification to lost item owner
|
||||
if err := w.sendMatchNotification(&match); err != nil {
|
||||
log.Printf("❌ Failed to send notification for match ID %d: %v", match.ID, err)
|
||||
continue
|
||||
}
|
||||
|
||||
// Mark as notified
|
||||
if err := w.matchRepo.MarkAsNotified(match.ID); err != nil {
|
||||
log.Printf("❌ Failed to mark match ID %d as notified: %v", match.ID, err)
|
||||
continue
|
||||
}
|
||||
|
||||
sentCount++
|
||||
}
|
||||
|
||||
log.Printf("✅ Sent %d notifications", sentCount)
|
||||
}
|
||||
|
||||
// sendMatchNotification sends a match notification
|
||||
func (w *NotificationWorker) sendMatchNotification(match *models.MatchResult) error {
|
||||
// Create notification
|
||||
err := models.CreateMatchNotification(
|
||||
w.db,
|
||||
match.LostItem.UserID,
|
||||
match.Item.Name,
|
||||
match.ID,
|
||||
)
|
||||
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
log.Printf("📧 Sent match notification to user ID %d for item: %s",
|
||||
match.LostItem.UserID, match.Item.Name)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// RunNow runs notification sending immediately (for testing)
|
||||
func (w *NotificationWorker) RunNow() {
|
||||
log.Println("▶️ Running notification sending manually...")
|
||||
w.sendPendingNotifications()
|
||||
}
|
||||
Reference in New Issue
Block a user