phuocntb/rikkei_simple_care
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

tam 2
All checks were successful
Deploy on Master Change / deploy (push) Successful in 1m25s
Details

Browse Source
This commit is contained in:
Nguyễn Phước
2026-07-13 09:04:09 +07:00
parent 719704f19e
commit 8803cd622a
22 changed files with 1322 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files

197
server/internal/websocket/websocket.go
View File

@@ -13,8 +13,11 @@ import (
)
const (
wsPingInterval = 30 * time.Second
wsPongWait = 60 * time.Second
// Heartbeat kiểu game: phát hiện mất kết nối trong ~30–45s.
wsPingInterval = 15 * time.Second
wsPongWait = 45 * time.Second
// Grace khi SV reconnect — tránh nhấp nháy online/offline.
studentOfflineGrace = 8 * time.Second
)
type SocketMsg struct {
@@ -44,6 +47,10 @@ func (c *SocketClient) WriteRaw(msg []byte) error {
return c.Conn.WriteMessage(websocket.TextMessage, msg)
}
type offlineGrace struct {
until time.Time
classID int64
}
type WsHub struct {
mu sync.RWMutex
@@ -51,6 +58,8 @@ type WsHub struct {
teachers map[string]*SocketClient
teachersByStaff map[uint][]string // staffId -> teacher connection addresses
subscribers map[int64][]string // studentId -> list of teacher connection addresses
grace map[int64]offlineGrace
graceTimers map[int64]*time.Timer
}
var Hub = &WsHub{
@@ -58,13 +67,18 @@ var Hub = &WsHub{
teachers: make(map[string]*SocketClient),
teachersByStaff: make(map[uint][]string),
subscribers: make(map[int64][]string),
grace: make(map[int64]offlineGrace),
graceTimers: make(map[int64]*time.Timer),
}
func (h *WsHub) IsStudentOnline(studentRkID int64) bool {
h.mu.RLock()
defer h.mu.RUnlock()
_, ok := h.students[studentRkID]
return ok
if _, ok := h.students[studentRkID]; ok {
return true
}
g, ok := h.grace[studentRkID]
return ok && time.Now().Before(g.until)
}
func (h *WsHub) PushExamToStudent(studentRkID int64, data map[string]any) {
@@ -102,16 +116,126 @@ func (h *WsHub) PushChatToStaff(staffID uint, data map[string]any) {
}
}
func (h *WsHub) BroadcastStudentViolation(data map[string]any) {
msg := SocketMsg{Event: "teacher:student-violation", Data: data}
msgBytes, err := json.Marshal(msg)
if err != nil {
return
}
h.mu.RLock()
defer h.mu.RUnlock()
for _, t := range h.teachers {
if t != nil {
_ = t.WriteRaw(msgBytes)
}
}
}
func (h *WsHub) broadcastPresence(studentID int64, online bool, classID int64) {
msg := SocketMsg{
Event: "presence:update",
Data: map[string]any{
"studentId": studentID,
"online": online,
"classId": classID,
},
}
msgBytes, err := json.Marshal(msg)
if err != nil {
return
}
for _, t := range h.teachers {
if t != nil {
_ = t.WriteRaw(msgBytes)
}
}
}
// ForceStudentOffline — tắt ngay (không chờ grace), dùng khi SV cố ý đóng app / vi phạm.
func (h *WsHub) ForceStudentOffline(studentID int64) {
h.mu.Lock()
defer h.mu.Unlock()
classID := int64(0)
if client, ok := h.students[studentID]; ok && client != nil {
classID = client.ClassID
delete(h.students, studentID)
}
if g, ok := h.grace[studentID]; ok {
if classID == 0 {
classID = g.classID
}
}
h.cancelGraceLocked(studentID)
log.Printf("[WS] Student %d force offline (intentional quit/violation)", studentID)
h.broadcastPresence(studentID, false, classID)
if teachers, exists := h.subscribers[studentID]; exists {
for _, tAddr := range teachers {
if t, found := h.teachers[tAddr]; found {
_ = t.WriteJSON(SocketMsg{
Event: "teacher:stream-stopped",
Data: map[string]any{"studentId": studentID},
})
}
}
}
}
func (h *WsHub) cancelGraceLocked(studentID int64) {
if t, ok := h.graceTimers[studentID]; ok {
t.Stop()
delete(h.graceTimers, studentID)
}
delete(h.grace, studentID)
}
func (h *WsHub) finalizeStudentOffline(studentID int64, classID int64) {
h.mu.Lock()
defer h.mu.Unlock()
if _, online := h.students[studentID]; online {
return
}
if _, ok := h.grace[studentID]; !ok {
return
}
delete(h.grace, studentID)
delete(h.graceTimers, studentID)
log.Printf("[WS] Student %d offline after grace", studentID)
h.broadcastPresence(studentID, false, classID)
if teachers, exists := h.subscribers[studentID]; exists {
for _, tAddr := range teachers {
if t, found := h.teachers[tAddr]; found {
_ = t.WriteJSON(SocketMsg{
Event: "teacher:stream-stopped",
Data: map[string]any{"studentId": studentID},
})
}
}
}
}
func (h *WsHub) GetOnlineStudentIDs(classID int64) []int64 {
h.mu.RLock()
defer h.mu.RUnlock()
seen := map[int64]bool{}
var ids []int64
now := time.Now()
for _, client := range h.students {
if client.ClassID == classID {
seen[client.StudentID] = true
ids = append(ids, client.StudentID)
}
}
for sid, g := range h.grace {
if now.Before(g.until) && g.classID == classID && !seen[sid] {
ids = append(ids, sid)
}
}
if ids == nil {
return []int64{}
}
@@ -124,8 +248,14 @@ func (h *WsHub) Register(c *SocketClient) {
c.Addr = c.Conn.RemoteAddr().String()
if c.Role == "student" {
wasInGrace := false
if _, ok := h.grace[c.StudentID]; ok {
wasInGrace = true
}
h.cancelGraceLocked(c.StudentID)
h.students[c.StudentID] = c
log.Printf("[WS] Student %d registered (Address: %s, Class: %d)", c.StudentID, c.Addr, c.ClassID)
log.Printf("[WS] Student %d registered (Address: %s, Class: %d, reconnectGrace=%v)", c.StudentID, c.Addr, c.ClassID, wasInGrace)
h.broadcastPresence(c.StudentID, true, c.ClassID)
if subs, exists := h.subscribers[c.StudentID]; exists && len(subs) > 0 {
_ = c.WriteJSON(SocketMsg{Event: "start_screenshot_stream"})
_ = c.WriteJSON(SocketMsg{Event: "start_webcam_stream"})
@@ -151,19 +281,17 @@ func (h *WsHub) Unregister(c *SocketClient) {
return
}
delete(h.students, c.StudentID)
log.Printf("[WS] Student %d disconnected", c.StudentID)
classID := c.ClassID
studentID := c.StudentID
// Báo cho các giáo viên đang xem là stream của học sinh đã dừng
if teachers, exists := h.subscribers[c.StudentID]; exists {
for _, tAddr := range teachers {
if t, found := h.teachers[tAddr]; found {
_ = t.WriteJSON(SocketMsg{
Event: "teacher:stream-stopped",
Data: map[string]any{"studentId": c.StudentID},
})
}
}
}
h.cancelGraceLocked(studentID)
deadline := time.Now().Add(studentOfflineGrace)
h.grace[studentID] = offlineGrace{until: deadline, classID: classID}
h.graceTimers[studentID] = time.AfterFunc(studentOfflineGrace, func() {
h.finalizeStudentOffline(studentID, classID)
})
log.Printf("[WS] Student %d disconnected — grace %v before offline", studentID, studentOfflineGrace)
// Chưa broadcast offline / stream-stopped — chờ grace (reconnect nhanh như game).
} else if c.Role == "teacher" {
delete(h.teachers, c.Addr)
if c.StaffID > 0 {
@@ -182,7 +310,6 @@ func (h *WsHub) Unregister(c *SocketClient) {
}
log.Printf("[WS] Teacher %s disconnected", c.Addr)
// Dọn dẹp subscriptions của giáo viên này
for sID, teachersList := range h.subscribers {
newList := []string{}
for _, addr := range teachersList {
@@ -192,7 +319,6 @@ func (h *WsHub) Unregister(c *SocketClient) {
}
if len(newList) == 0 {
delete(h.subscribers, sID)
// Nếu không còn ai xem học sinh này, gửi lệnh tắt camera/screen cho client học sinh
if student, exists := h.students[sID]; exists {
_ = student.WriteJSON(SocketMsg{Event: "stop_screenshot_stream"})
_ = student.WriteJSON(SocketMsg{Event: "stop_webcam_stream"})
@@ -204,12 +330,10 @@ func (h *WsHub) Unregister(c *SocketClient) {
}
}
// Teacher bắt đầu xem stream của Student
func (h *WsHub) Subscribe(teacherAddr string, studentID int64) {
h.mu.Lock()
defer h.mu.Unlock()
// Thêm giáo viên vào list người xem của học sinh
teachersList := h.subscribers[studentID]
alreadySubscribed := false
for _, addr := range teachersList {
@@ -223,14 +347,12 @@ func (h *WsHub) Subscribe(teacherAddr string, studentID int64) {
log.Printf("[WS] Teacher %s subscribed to student %d stream", teacherAddr, studentID)
}
// Phát lệnh cho máy học sinh bật stream (nếu học sinh đang online)
if student, exists := h.students[studentID]; exists {
_ = student.WriteJSON(SocketMsg{Event: "start_screenshot_stream"})
_ = student.WriteJSON(SocketMsg{Event: "start_webcam_stream"})
}
}
// Teacher dừng xem stream của Student
func (h *WsHub) Unsubscribe(teacherAddr string, studentID int64) {
h.mu.Lock()
defer h.mu.Unlock()
@@ -250,8 +372,6 @@ func (h *WsHub) Unsubscribe(teacherAddr string, studentID int64) {
if len(newList) == 0 {
delete(h.subscribers, studentID)
log.Printf("[WS] Student %d has no more proctor subscribers. Stopping streams.", studentID)
// Báo học sinh tắt camera & screen stream để tiết kiệm mạng và CPU
if student, exists := h.students[studentID]; exists {
_ = student.WriteJSON(SocketMsg{Event: "stop_screenshot_stream"})
_ = student.WriteJSON(SocketMsg{Event: "stop_webcam_stream"})
@@ -261,7 +381,6 @@ func (h *WsHub) Unsubscribe(teacherAddr string, studentID int64) {
}
}
// Chuyển tiếp frame ảnh từ Student đến các Teacher đã subscribe
func (h *WsHub) RelayFrame(studentID int64, event string, data map[string]any) {
h.mu.RLock()
teachersList, exists := h.subscribers[studentID]
@@ -269,7 +388,6 @@ func (h *WsHub) RelayFrame(studentID int64, event string, data map[string]any) {
h.mu.RUnlock()
return
}
// Snapshot subscriber addresses while holding read lock
addrs := make([]string, len(teachersList))
copy(addrs, teachersList)
h.mu.RUnlock()
@@ -291,18 +409,30 @@ func (h *WsHub) RelayFrame(studentID int64, event string, data map[string]any) {
return
}
var dead []string
h.mu.RLock()
defer h.mu.RUnlock()
for _, addr := range addrs {
if t, found := h.teachers[addr]; found {
if err := t.WriteRaw(msgBytes); err != nil {
log.Printf("[WS] Relay to teacher %s failed: %v", addr, err)
dead = append(dead, addr)
}
}
}
h.mu.RUnlock()
for _, addr := range dead {
if t, ok := func() (*SocketClient, bool) {
h.mu.RLock()
defer h.mu.RUnlock()
t, ok := h.teachers[addr]
return t, ok
}(); ok && t != nil {
_ = t.Conn.Close()
}
}
}
// WebSocket handler cho Fiber route
func WebSocketHandler(db *gorm.DB) func(*websocket.Conn) {
return func(c *websocket.Conn) {
role := c.Query("role", "student")
@@ -351,7 +481,7 @@ func WebSocketHandler(db *gorm.DB) func(*websocket.Conn) {
return
case <-ticker.C:
client.writeMu.Lock()
err := c.WriteMessage(websocket.PingMessage, nil)
err := c.WriteControl(websocket.PingMessage, []byte("ping"), time.Now().Add(5*time.Second))
client.writeMu.Unlock()
if err != nil {
return
@@ -372,14 +502,14 @@ func WebSocketHandler(db *gorm.DB) func(*websocket.Conn) {
continue
}
// Xử lý các sự kiện
switch msg.Event {
case "client:ping":
_ = client.WriteJSON(SocketMsg{Event: "client:pong", Data: map[string]any{"t": time.Now().UnixMilli()}})
case "screenshot_stream_frame", "webcam_stream_frame":
// Nhận frame từ học sinh, chuyển tiếp về các thầy cô
Hub.RelayFrame(client.StudentID, msg.Event, msg.Data)
case "teacher:subscribe":
// Giáo viên đăng ký xem học sinh cụ thể
if client.Role == "teacher" {
if sIDVal, ok := msg.Data["studentId"]; ok {
var sID int64
@@ -396,7 +526,6 @@ func WebSocketHandler(db *gorm.DB) func(*websocket.Conn) {
}
case "teacher:unsubscribe":
// Giáo viên hủy đăng ký
if client.Role == "teacher" {
if sIDVal, ok := msg.Data["studentId"]; ok {
var sID int64