Add assignment engine, FCM, WebSockets, city gate, and internal APIs
- internal/assignment: GEORADIUS miler assignment with retry/escalation, customer-side provider scoring, FCM notifications on assign - internal/notify: Firebase Admin SDK (FCM) client initialisation - internal/ws: WebSocket handlers for live parcel tracking and customer↔miler chat - middlewares: city gate (pincode prefix validation), internal API key auth, WebSocket JWT auth - controllers: InternalNotify + InternalReassign for machine-to-machine calls; pricing helpers wired into CreateCustomerBooking and CreateCRMBooking - routes: /internal/*, /ws/bookings/:id/track, /ws/bookings/:id/chat - models/users, models/doormile_pricing: new fields for device tokens, assignment state, pricing bands - seed_data.sql: initial pricing seed rows .env and Firebase service-account JSON intentionally excluded. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
271
internal/ws/chat.go
Normal file
271
internal/ws/chat.go
Normal file
@@ -0,0 +1,271 @@
|
||||
package ws
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"doormile/db"
|
||||
"doormile/models"
|
||||
"doormile/utils"
|
||||
|
||||
"github.com/gofiber/websocket/v2"
|
||||
)
|
||||
|
||||
// chatMessage is the JSON frame broadcast to the receiving participant.
|
||||
type chatMessage struct {
|
||||
Sender string `json:"sender"`
|
||||
Message string `json:"message"`
|
||||
Timestamp string `json:"timestamp"`
|
||||
}
|
||||
|
||||
// chatConn wraps a single WebSocket connection with a write mutex so concurrent
|
||||
// senders (broadcast from the other peer, poller close) never interleave frames.
|
||||
type chatConn struct {
|
||||
conn *websocket.Conn
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
func (cc *chatConn) send(data []byte) {
|
||||
cc.mu.Lock()
|
||||
defer cc.mu.Unlock()
|
||||
cc.conn.WriteMessage(websocket.TextMessage, data) //nolint:errcheck — best-effort delivery
|
||||
}
|
||||
|
||||
func (cc *chatConn) close() {
|
||||
cc.mu.Lock()
|
||||
defer cc.mu.Unlock()
|
||||
cc.conn.Close() //nolint:errcheck
|
||||
}
|
||||
|
||||
// chatRoom holds at most two named slots: "customer" and "miler".
|
||||
// All slot mutations are protected by mu.
|
||||
type chatRoom struct {
|
||||
bookingID int
|
||||
mu sync.Mutex
|
||||
slots map[string]*chatConn
|
||||
closeOnce sync.Once // ensures shutdown() body runs exactly once
|
||||
closeCh chan struct{} // closed by shutdown()
|
||||
pollerOnce sync.Once // ensures exactly one status-poller goroutine per room
|
||||
}
|
||||
|
||||
// rooms is the process-wide registry of active chat rooms, keyed by bookingID.
|
||||
var rooms sync.Map // map[int]*chatRoom
|
||||
|
||||
// ChatHandler is the WebSocket handler for the ephemeral per-booking chat room.
|
||||
//
|
||||
// Route: GET /ws/bookings/:bookingid/chat
|
||||
// Params: role=customer|miler (query)
|
||||
// token=<JWT> (query, validated by WsChatAuth middleware)
|
||||
//
|
||||
// At most two participants (one customer, one miler) may occupy a room.
|
||||
// Messages received from one participant are forwarded to the other.
|
||||
// The room closes automatically when the booking reaches a terminal status.
|
||||
func ChatHandler(c *websocket.Conn) {
|
||||
bookingID, err := strconv.Atoi(c.Params("bookingid"))
|
||||
if err != nil {
|
||||
sendError(c, "invalid booking ID")
|
||||
return
|
||||
}
|
||||
|
||||
role := c.Query("role")
|
||||
if role != "customer" && role != "miler" {
|
||||
sendError(c, "role must be 'customer' or 'miler'")
|
||||
return
|
||||
}
|
||||
|
||||
// Verify the booking exists before letting anyone into the room.
|
||||
var booking models.PickupBooking
|
||||
if err := db.DB.First(&booking, bookingID).Error; err != nil {
|
||||
sendError(c, "booking not found")
|
||||
return
|
||||
}
|
||||
if isTerminalStatus(booking.Status) {
|
||||
sendError(c, "chat is closed — booking is already completed")
|
||||
return
|
||||
}
|
||||
|
||||
room, err := joinRoom(bookingID, role, c)
|
||||
if err != nil {
|
||||
sendError(c, err.Error())
|
||||
return
|
||||
}
|
||||
defer room.leave(role)
|
||||
|
||||
utils.Info("WS/Chat: participant joined", "booking_id", bookingID, "role", role)
|
||||
|
||||
// Block here reading client messages. Exits on client disconnect or room close.
|
||||
for {
|
||||
_, raw, err := c.ReadMessage()
|
||||
if err != nil {
|
||||
break
|
||||
}
|
||||
room.broadcast(role, string(raw))
|
||||
}
|
||||
|
||||
utils.Info("WS/Chat: participant left", "booking_id", bookingID, "role", role)
|
||||
}
|
||||
|
||||
// ─── Room lifecycle ───────────────────────────────────────────────────────────
|
||||
|
||||
// joinRoom finds or creates the room for bookingID, registers the connection in
|
||||
// the given role slot, and starts the booking-status poller on first join.
|
||||
// Returns an error if the slot is already occupied or the room is shutting down.
|
||||
func joinRoom(bookingID int, role string, c *websocket.Conn) (*chatRoom, error) {
|
||||
val, _ := rooms.LoadOrStore(bookingID, &chatRoom{
|
||||
bookingID: bookingID,
|
||||
slots: make(map[string]*chatConn),
|
||||
closeCh: make(chan struct{}),
|
||||
})
|
||||
room := val.(*chatRoom)
|
||||
|
||||
room.mu.Lock()
|
||||
defer room.mu.Unlock()
|
||||
|
||||
// If shutdown already started (e.g., race with a closing poller), reject.
|
||||
select {
|
||||
case <-room.closeCh:
|
||||
return nil, fmt.Errorf("chat room is closed")
|
||||
default:
|
||||
}
|
||||
|
||||
if _, exists := room.slots[role]; exists {
|
||||
return nil, fmt.Errorf("%s is already connected to this room", role)
|
||||
}
|
||||
|
||||
room.slots[role] = &chatConn{conn: c}
|
||||
|
||||
// Start the status poller the first time any participant joins.
|
||||
room.pollerOnce.Do(func() { go room.pollBookingStatus() })
|
||||
|
||||
return room, nil
|
||||
}
|
||||
|
||||
// leave removes the participant from the room and shuts the room down if empty.
|
||||
// Called via defer in ChatHandler — safe even after a force-close.
|
||||
func (r *chatRoom) leave(role string) {
|
||||
r.mu.Lock()
|
||||
delete(r.slots, role)
|
||||
isEmpty := len(r.slots) == 0
|
||||
r.mu.Unlock()
|
||||
|
||||
if isEmpty {
|
||||
r.shutdown()
|
||||
}
|
||||
}
|
||||
|
||||
// broadcast wraps the raw text in a chatMessage frame and sends it to every
|
||||
// participant other than the sender.
|
||||
func (r *chatRoom) broadcast(senderRole, text string) {
|
||||
frame := chatMessage{
|
||||
Sender: senderRole,
|
||||
Message: text,
|
||||
Timestamp: time.Now().UTC().Format(time.RFC3339),
|
||||
}
|
||||
data, err := json.Marshal(frame)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
r.mu.Lock()
|
||||
targets := make([]*chatConn, 0, 1)
|
||||
for role, cc := range r.slots {
|
||||
if role != senderRole {
|
||||
targets = append(targets, cc)
|
||||
}
|
||||
}
|
||||
r.mu.Unlock()
|
||||
|
||||
for _, cc := range targets {
|
||||
cc.send(data)
|
||||
}
|
||||
}
|
||||
|
||||
// shutdown closes the done channel and removes the room from the global registry.
|
||||
// Protected by closeOnce — safe to call from both the poller and connection handlers.
|
||||
func (r *chatRoom) shutdown() {
|
||||
r.closeOnce.Do(func() {
|
||||
close(r.closeCh)
|
||||
rooms.Delete(r.bookingID)
|
||||
utils.Info("WS/Chat: room removed from registry", "booking_id", r.bookingID)
|
||||
})
|
||||
}
|
||||
|
||||
// forceClose closes every open WebSocket connection then shuts the room down.
|
||||
// Called by the status poller when the booking reaches a terminal status.
|
||||
// Each handler's c.ReadMessage() will return an error, causing it to break
|
||||
// its read loop and call room.leave() — which hits shutdown() again (no-op).
|
||||
func (r *chatRoom) forceClose() {
|
||||
r.mu.Lock()
|
||||
conns := make([]*chatConn, 0, len(r.slots))
|
||||
for _, cc := range r.slots {
|
||||
conns = append(conns, cc)
|
||||
}
|
||||
r.mu.Unlock()
|
||||
|
||||
for _, cc := range conns {
|
||||
cc.close()
|
||||
}
|
||||
|
||||
r.shutdown()
|
||||
}
|
||||
|
||||
// ─── Status poller ────────────────────────────────────────────────────────────
|
||||
|
||||
// pollBookingStatus runs in a goroutine for the room's lifetime.
|
||||
// Every 5 s it reloads the booking; when it hits a terminal status it publishes
|
||||
// the NATS event, force-closes all connections, and exits.
|
||||
func (r *chatRoom) pollBookingStatus() {
|
||||
t := time.NewTicker(5 * time.Second)
|
||||
defer t.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-r.closeCh:
|
||||
return
|
||||
|
||||
case <-t.C:
|
||||
var booking models.PickupBooking
|
||||
if err := db.DB.First(&booking, r.bookingID).Error; err != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
if isTerminalStatus(booking.Status) {
|
||||
utils.Info("WS/Chat: closing room — booking reached terminal status",
|
||||
"booking_id", r.bookingID,
|
||||
"status", booking.Status,
|
||||
)
|
||||
publishChatRoomClosed(r.bookingID)
|
||||
r.forceClose()
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// publishChatRoomClosed publishes a NATS event so downstream services know the
|
||||
// chat session has ended. Non-fatal — logs a warning on failure.
|
||||
func publishChatRoomClosed(bookingID int) {
|
||||
if db.Js == nil {
|
||||
return
|
||||
}
|
||||
|
||||
payload := map[string]interface{}{
|
||||
"booking_id": bookingID,
|
||||
"closed_at": time.Now().UnixMilli(),
|
||||
}
|
||||
data, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
subject := fmt.Sprintf("chat.room.closed.%d", bookingID)
|
||||
if _, err := db.Js.Publish(subject, data); err != nil {
|
||||
utils.Warn("WS/Chat: failed to publish room-closed event",
|
||||
"booking_id", bookingID,
|
||||
"error", err,
|
||||
)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user