grpccanary/lib/grpc/modular/server.go

package modular

import (
	"context"
	"fmt"
	"io"
	"math/rand"
	"net"
	"sync"
	"time"

	"grpccanary/lib/grpc/modular/protoapi"

	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/reflection"
	"google.golang.org/grpc/status"
)

type UploadedFile struct {
	FileName   string
	Content    []byte
	UploadedAt int64
}

type AlertSubscriber struct {
	ClientId string
	Channel  chan *protoapi.AlertMessage
}

var (
	fileStore   = make(map[string]*UploadedFile)
	storeMu     sync.RWMutex
	subscribers = make(map[string]*AlertSubscriber)
	subMu       sync.Mutex
)

func publishAlert(alert *protoapi.AlertMessage) {
	subMu.Lock()
	defer subMu.Unlock()
	for _, sub := range subscribers {
		select {
		case sub.Channel <- alert:
		default:
			fmt.Printf("Alert channel blocked for client %s, dropping event\n", sub.ClientId)
		}
	}
}

func random(min, max int, src rand.Source) int {
	return rand.New(src).Intn(max-min) + min
}

func getString(len int64, src rand.Source) string {
	temp := ""
	startChar := "!"
	var i int64 = 1
	for {
		myRand := random(0, 94, src)
		newChar := string(startChar[0] + byte(myRand))
		temp = temp + newChar
		if i == len {
			break
		}
		i++
	}
	return temp
}

// 1. Core Server Implementation
type CoreServer struct {
	protoapi.UnimplementedCoreServiceServer
}

func (CoreServer) GetDate(ctx context.Context, r *protoapi.RequestDateTime) (*protoapi.DateTime, error) {
	currentTime := time.Now()
	return &protoapi.DateTime{
		Value: currentTime.String(),
	}, nil
}

func (CoreServer) UpdateSensingData(ctx context.Context, r *protoapi.SensingData) (*protoapi.SensingResponse, error) {
	fmt.Printf("[Modular Core] Received sensing data - Device: %s, Temp: %.2f°C, Humid: %.2f%%\n", r.GetDeviceId(), r.GetTemperature(), r.GetHumidity())

	if r.GetTemperature() > 40.0 {
		fmt.Printf("⚠️ [Modular Core] Critical temperature detected: %.2f°C! Publishing warning...\n", r.GetTemperature())
		publishAlert(&protoapi.AlertMessage{
			AlertId:   fmt.Sprintf("alert-%d", time.Now().UnixNano()),
			DeviceId:  r.GetDeviceId(),
			Message:   fmt.Sprintf("Critical high temperature: %.2f°C (Humidity: %.2f%%)", r.GetTemperature(), r.GetHumidity()),
			Timestamp: time.Now().Unix(),
		})
	}

	return &protoapi.SensingResponse{
		Success: true,
		Message: fmt.Sprintf("Sensing data updated successfully for device %s", r.GetDeviceId()),
	}, nil
}

func (CoreServer) GetRandomPass(ctx context.Context, r *protoapi.RequestPass) (*protoapi.RandomPass, error) {
	src := rand.NewSource(r.GetSeed())
	temp := getString(r.GetLength(), src)
	return &protoapi.RandomPass{
		Password: temp,
	}, nil
}

// 2. File Transfer Server Implementation
type FileTransferServer struct {
	protoapi.UnimplementedFileTransferServiceServer
}

func (FileTransferServer) UploadFile(stream protoapi.FileTransferService_UploadFileServer) error {
	var totalBytes int64
	var fileName string
	var buffer []byte

	for {
		chunk, err := stream.Recv()
		if err == io.EOF {
			fmt.Printf("[Modular File] File upload completed. Received %d bytes for file '%s'\n", totalBytes, fileName)

			if fileName != "" {
				storeMu.Lock()
				fileStore[fileName] = &UploadedFile{
					FileName:   fileName,
					Content:    buffer,
					UploadedAt: time.Now().Unix(),
				}
				storeMu.Unlock()
			}

			return stream.SendAndClose(&protoapi.UploadStatus{
				Success:       true,
				Message:       fmt.Sprintf("File '%s' uploaded successfully.", fileName),
				BytesUploaded: totalBytes,
			})
		}
		if err != nil {
			fmt.Println("[Modular File] File upload error:", err)
			return err
		}

		if fileName == "" {
			fileName = chunk.GetFileName()
		}
		buffer = append(buffer, chunk.GetContent()...)
		totalBytes += int64(len(chunk.GetContent()))
	}
}

func (FileTransferServer) ListFiles(ctx context.Context, r *protoapi.EmptyRequest) (*protoapi.FileList, error) {
	storeMu.RLock()
	defer storeMu.RUnlock()

	var files []*protoapi.FileMetadata
	for _, f := range fileStore {
		files = append(files, &protoapi.FileMetadata{
			FileName:   f.FileName,
			FileSize:   int64(len(f.Content)),
			UploadedAt: f.UploadedAt,
		})
	}

	return &protoapi.FileList{Files: files}, nil
}

func (FileTransferServer) DownloadFile(r *protoapi.DownloadRequest, stream protoapi.FileTransferService_DownloadFileServer) error {
	storeMu.RLock()
	f, exists := fileStore[r.GetFileName()]
	storeMu.RUnlock()

	if !exists {
		return status.Errorf(codes.NotFound, "file %s not found", r.GetFileName())
	}

	chunkSize := 1024
	totalBytes := len(f.Content)

	for i := 0; i < totalBytes; i += chunkSize {
		end := i + chunkSize
		if end > totalBytes {
			end = totalBytes
		}

		err := stream.Send(&protoapi.FileChunk{
			FileName: f.FileName,
			Content:  f.Content[i:end],
		})
		if err != nil {
			return err
		}
	}

	return nil
}

// 3. Alert Server Implementation
type AlertServer struct {
	protoapi.UnimplementedAlertServiceServer
}

func (AlertServer) SubscribeAlerts(r *protoapi.AlertSubscription, stream protoapi.AlertService_SubscribeAlertsServer) error {
	clientId := r.GetClientId()
	ch := make(chan *protoapi.AlertMessage, 10)
	sub := &AlertSubscriber{
		ClientId: clientId,
		Channel:  ch,
	}

	subMu.Lock()
	subscribers[clientId] = sub
	subMu.Unlock()

	fmt.Printf("[Modular Alert] Client %s subscribed to alerts on topic '%s'\n", clientId, r.GetTopic())

	for {
		select {
		case alert := <-ch:
			err := stream.Send(alert)
			if err != nil {
				subMu.Lock()
				delete(subscribers, clientId)
				subMu.Unlock()
				fmt.Printf("[Modular Alert] Client %s alert subscription disconnected: %v\n", clientId, err)
				return err
			}
		case <-stream.Context().Done():
			subMu.Lock()
			delete(subscribers, clientId)
			subMu.Unlock()
			fmt.Printf("[Modular Alert] Client %s unsubscribed (context done)\n", clientId)
			return nil
		}
	}
}

func ServerRun(addr string) {
	server := grpc.NewServer()

	// Register multiple services on the same gRPC server
	protoapi.RegisterCoreServiceServer(server, CoreServer{})
	protoapi.RegisterFileTransferServiceServer(server, FileTransferServer{})
	protoapi.RegisterAlertServiceServer(server, AlertServer{})

	reflection.Register(server)

	listen, err := net.Listen("tcp", addr)
	if err != nil {
		fmt.Println(err)
		return
	}

	fmt.Printf("[Modular Server] Listening on %s...\n", addr)
	server.Serve(listen)
}