Task Management Platform

package main

import (
    "context"
    "fmt"
    "log"
    "time"

    "github.com/go-redis/redis/v8"
    "google.golang.org/grpc"
    "gorm.io/gorm"
)

type TaskService struct {
    pb.UnimplementedTaskServiceServer
    db           *gorm.DB
    redis        *redis.Client
    aiService    *AIService
    kafkaClient  *kafka.Writer
}

// CreateTask handles task creation with AI-powered optimization
func (s *TaskService) CreateTask(ctx context.Context, req *pb.CreateTaskRequest) (*pb.Task, error) {
    // Validate request
    if err := s.validateTaskRequest(req); err != nil {
        return nil, status.Errorf(codes.InvalidArgument, "validation failed: %v", err)
    }
    
    // Create task entity
    task := &Task{
        ID:          generateUUID(),
        Title:       req.Title,
        Description: req.Description,
        ProjectID:   req.ProjectId,
        AssigneeID:  req.AssigneeId,
        Priority:    req.Priority,
        Status:      TaskStatusPending,
        CreatedAt:   time.Now(),
        UpdatedAt:   time.Now(),
    }
    
    // AI-powered task optimization
    aiInsights, err := s.aiService.AnalyzeTask(ctx, task)
    if err != nil {
        log.Printf("AI analysis failed: %v", err)
    } else {
        task.EstimatedHours = aiInsights.EstimatedHours
        task.SuggestedPriority = aiInsights.SuggestedPriority
        task.OptimalAssignee = aiInsights.OptimalAssignee
    }
    
    // Database transaction
    tx := s.db.Begin()
    defer func() {
        if r := recover(); r != nil {
            tx.Rollback()
        }
    }()
    
    if err := tx.Create(task).Error; err != nil {
        tx.Rollback()
        return nil, status.Errorf(codes.Internal, "failed to create task: %v", err)
    }
    
    // Cache the task
    taskJSON, _ := json.Marshal(task)
    s.redis.Set(ctx, fmt.Sprintf("task:%s", task.ID), taskJSON, time.Hour*24)
    
    // Publish event to Kafka
    event := &TaskCreatedEvent{
        TaskID:    task.ID,
        ProjectID: task.ProjectID,
        AssigneeID: task.AssigneeID,
        Timestamp: time.Now(),
    }
    
    if err := s.publishEvent(ctx, "task.created", event); err != nil {
        log.Printf("Failed to publish event: %v", err)
    }
    
    tx.Commit()
    
    return task.ToProto(), nil
}

// UpdateTaskStatus with real-time notifications
func (s *TaskService) UpdateTaskStatus(ctx context.Context, req *pb.UpdateTaskStatusRequest) (*pb.Task, error) {
    // Get task from cache first
    taskKey := fmt.Sprintf("task:%s", req.TaskId)
    cached, err := s.redis.Get(ctx, taskKey).Result()
    
    var task *Task
    if err == redis.Nil {
        // Not in cache, fetch from database
        if err := s.db.First(&task, "id = ?", req.TaskId).Error; err != nil {
            return nil, status.Errorf(codes.NotFound, "task not found")
        }
    } else {
        json.Unmarshal([]byte(cached), &task)
    }
    
    // Update status with validation
    oldStatus := task.Status
    task.Status = req.NewStatus
    task.UpdatedAt = time.Now()
    
    // AI-powered status change validation
    if valid, reason := s.aiService.ValidateStatusChange(ctx, oldStatus, req.NewStatus, task); !valid {
        return nil, status.Errorf(codes.FailedPrecondition, "invalid status change: %s", reason)
    }
    
    // Update in database
    if err := s.db.Save(task).Error; err != nil {
        return nil, status.Errorf(codes.Internal, "failed to update task: %v", err)
    }
    
    // Update cache
    taskJSON, _ := json.Marshal(task)
    s.redis.Set(ctx, taskKey, taskJSON, time.Hour*24)
    
    // Publish real-time update
    statusEvent := &TaskStatusChangedEvent{
        TaskID:    task.ID,
        OldStatus: oldStatus,
        NewStatus: req.NewStatus,
        UpdatedBy: getUserFromContext(ctx),
        Timestamp: time.Now(),
    }
    
    s.publishEvent(ctx, "task.status.changed", statusEvent)
    
    // Trigger automation workflows
    go s.triggerAutomationWorkflows(task, oldStatus, req.NewStatus)
    
    return task.ToProto(), nil
}

// AI-powered task assignment optimization
func (s *TaskService) OptimizeTaskAssignment(ctx context.Context, req *pb.OptimizeAssignmentRequest) (*pb.AssignmentRecommendations, error) {
    // Get team workload data
    workloadData, err := s.getTeamWorkloadData(ctx, req.TeamId)
    if err != nil {
        return nil, status.Errorf(codes.Internal, "failed to get workload data: %v", err)
    }
    
    // Get task complexity analysis
    tasks, err := s.getUnassignedTasks(ctx, req.ProjectId)
    if err != nil {
        return nil, status.Errorf(codes.Internal, "failed to get tasks: %v", err)
    }
    
    // AI optimization
    recommendations, err := s.aiService.OptimizeAssignments(ctx, &AIOptimizationRequest{
        Tasks:        tasks,
        TeamMembers:  workloadData.Members,
        Constraints:  req.Constraints,
        Objectives:   req.Objectives,
    })
    
    if err != nil {
        return nil, status.Errorf(codes.Internal, "AI optimization failed: %v", err)
    }
    
    return &pb.AssignmentRecommendations{
        Recommendations: recommendations.ToProto(),
        ConfidenceScore: recommendations.ConfidenceScore,
        ExpectedImprovement: recommendations.ExpectedImprovement,
    }, nil
}

type AIService struct {
    mlClient    *MLClient
    modelCache  *ModelCache
}

func (ai *AIService) AnalyzeTask(ctx context.Context, task *Task) (*AITaskInsights, error) {
    // Feature extraction
    features := ai.extractTaskFeatures(task)
    
    // Load trained models
    estimationModel, err := ai.modelCache.GetModel("task_estimation")
    if err != nil {
        return nil, err
    }
    
    priorityModel, err := ai.modelCache.GetModel("priority_prediction")
    if err != nil {
        return nil, err
    }
    
    // Run predictions
    estimatedHours := estimationModel.Predict(features.EstimationFeatures)
    suggestedPriority := priorityModel.Predict(features.PriorityFeatures)
    
    // Optimal assignee prediction
    optimalAssignee, err := ai.predictOptimalAssignee(ctx, task, features)
    if err != nil {
        log.Printf("Assignee prediction failed: %v", err)
    }
    
    return &AITaskInsights{
        EstimatedHours:    estimatedHours,
        SuggestedPriority: suggestedPriority,
        OptimalAssignee:   optimalAssignee,
        ConfidenceScore:   ai.calculateConfidence(features),
    }, nil
}

func (ai *AIService) OptimizeAssignments(ctx context.Context, req *AIOptimizationRequest) (*OptimizationResult, error) {
    // Multi-objective optimization using genetic algorithm
    optimizer := &GeneticOptimizer{
        PopulationSize: 100,
        Generations:    50,
        MutationRate:   0.1,
        CrossoverRate:  0.8,
    }
    
    // Define fitness function
    fitnessFunc := func(assignment []int) float64 {
        return ai.calculateAssignmentFitness(assignment, req)
    }
    
    // Run optimization
    bestAssignment, score := optimizer.Optimize(fitnessFunc, len(req.Tasks), len(req.TeamMembers))
    
    // Convert to recommendations
    recommendations := make([]*AssignmentRecommendation, len(req.Tasks))
    for i, taskIdx := range bestAssignment {
        recommendations[i] = &AssignmentRecommendation{
            TaskID:           req.Tasks[taskIdx].ID,
            RecommendedAssignee: req.TeamMembers[i].ID,
            ConfidenceScore:  score,
            Reasoning:        ai.generateReasoningText(req.Tasks[taskIdx], req.TeamMembers[i]),
        }
    }
    
    return &OptimizationResult{
        Recommendations:     recommendations,
        ConfidenceScore:     score,
        ExpectedImprovement: ai.calculateExpectedImprovement(req, recommendations),
    }, nil
}

package websocket

import (
    "encoding/json"
    "log"
    "net/http"
    "sync"
    
    "github.com/gorilla/websocket"
)

type Hub struct {
    clients     map[*Client]bool
    broadcast   chan []byte
    register    chan *Client
    unregister  chan *Client
    rooms       map[string]map[*Client]bool
    roomsMutex  sync.RWMutex
}

type Client struct {
    hub      *Hub
    conn     *websocket.Conn
    send     chan []byte
    userID   string
    projectID string
}

type Message struct {
    Type      string      `json:"type"`
    ProjectID string      `json:"project_id"`
    UserID    string      `json:"user_id"`
    Data      interface{} `json:"data"`
    Timestamp int64       `json:"timestamp"`
}

func NewHub() *Hub {
    return &Hub{
        clients:    make(map[*Client]bool),
        broadcast:  make(chan []byte),
        register:   make(chan *Client),
        unregister: make(chan *Client),
        rooms:      make(map[string]map[*Client]bool),
    }
}

func (h *Hub) Run() {
    for {
        select {
        case client := <-h.register:
            h.registerClient(client)
            
        case client := <-h.unregister:
            h.unregisterClient(client)
            
        case message := <-h.broadcast:
            h.broadcastMessage(message)
        }
    }
}

func (h *Hub) registerClient(client *Client) {
    h.clients[client] = true
    
    // Add to project room
    h.roomsMutex.Lock()
    if h.rooms[client.projectID] == nil {
        h.rooms[client.projectID] = make(map[*Client]bool)
    }
    h.rooms[client.projectID][client] = true
    h.roomsMutex.Unlock()
    
    // Send welcome message
    welcomeMsg := &Message{
        Type:      "user_joined",
        ProjectID: client.projectID,
        UserID:    client.userID,
        Data: map[string]interface{}{
            "message": "User joined the project",
        },
        Timestamp: time.Now().Unix(),
    }
    
    h.broadcastToRoom(client.projectID, welcomeMsg)
    log.Printf("Client %s joined project %s", client.userID, client.projectID)
}

func (h *Hub) broadcastToRoom(projectID string, message *Message) {
    h.roomsMutex.RLock()
    room := h.rooms[projectID]
    h.roomsMutex.RUnlock()
    
    if room == nil {
        return
    }
    
    messageBytes, err := json.Marshal(message)
    if err != nil {
        log.Printf("Error marshaling message: %v", err)
        return
    }
    
    for client := range room {
        select {
        case client.send <- messageBytes:
        default:
            // Client buffer full, remove client
            h.unregisterClient(client)
        }
    }
}

// Handle real-time task updates
func (h *Hub) HandleTaskUpdate(taskUpdate *TaskUpdateEvent) {
    message := &Message{
        Type:      "task_updated",
        ProjectID: taskUpdate.ProjectID,
        UserID:    taskUpdate.UpdatedBy,
        Data: map[string]interface{}{
            "task_id":     taskUpdate.TaskID,
            "field":       taskUpdate.Field,
            "old_value":   taskUpdate.OldValue,
            "new_value":   taskUpdate.NewValue,
            "timestamp":   taskUpdate.Timestamp,
        },
        Timestamp: time.Now().Unix(),
    }
    
    h.broadcastToRoom(taskUpdate.ProjectID, message)
}

// Handle collaborative editing
func (h *Hub) HandleCollaborativeEdit(edit *CollaborativeEditEvent) {
    message := &Message{
        Type:      "collaborative_edit",
        ProjectID: edit.ProjectID,
        UserID:    edit.UserID,
        Data: map[string]interface{}{
            "task_id":   edit.TaskID,
            "field":     edit.Field,
            "operation": edit.Operation,
            "position":  edit.Position,
            "content":   edit.Content,
            "version":   edit.Version,
        },
        Timestamp: time.Now().Unix(),
    }
    
    // Apply operational transformation for conflict resolution
    resolvedEdit := h.applyOperationalTransform(edit)
    if resolvedEdit != nil {
        message.Data = resolvedEdit
    }
    
    h.broadcastToRoom(edit.ProjectID, message)
}

func (h *Hub) applyOperationalTransform(edit *CollaborativeEditEvent) map[string]interface{} {
    // Implement operational transformation algorithm
    // for real-time collaborative editing without conflicts
    
    // This is a simplified version - production would need
    // more sophisticated conflict resolution
    return map[string]interface{}{
        "task_id":   edit.TaskID,
        "field":     edit.Field,
        "operation": edit.Operation,
        "position":  edit.Position,
        "content":   edit.Content,
        "version":   edit.Version + 1, // Increment version
        "resolved":  true,
    }
}