admin/go-socket
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add hub event sending

Browse Source
This commit is contained in:
cos
2026-05-08 11:11:28 +02:00
parent 40acd753ac
commit 53652d1f79
7 changed files with 330 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/httpRequest/hubs.go
+193 -41
View File
@@ -124,6 +124,27 @@ func haveHubUserCachedPermissions(needed types.CachedUserPermissions, user *type
return true
}
func sendChannelUpdatesToUser(target *types.User, hub *types.Hub) {
hub.Mu.RLock()
channels := make([]*types.HubChannel, 0, len(hub.Channels))
for _, ch := range hub.Channels {
channels = append(channels, ch)
}
hub.Mu.RUnlock()
for _, ch := range channels {
ch.Mu.RLock()
perm, ok := ch.UsersCachedPermissions[target.Id]
ch.Mu.RUnlock()
if ok && perm.CanView() {
wsServer.WsSendMessageCloseIfTimeout(target, &types.WsHubSpecificHubEventMessage{
Type: WsEventType.HubChannelUpdate,
HubId: hub.Id,
Event: ch,
})
}
}
}
func HandleHubCreate(response http.ResponseWriter, request *http.Request) {
if !validCheckWithResponseOnFail(response, request, normal) {
return
@@ -258,6 +279,13 @@ func HandleHubJoin(response http.ResponseWriter, request *http.Request) {
return
}
response.WriteHeader(http.StatusCreated)
wsServer.WsSendEventMessageToHubUsersCloseIfTimeout(user.Id, hub, &types.WsHubSpecificHubEventMessage{
Type: WsEventType.HubUserJoin,
Event: &map[string]any{
"userId": user.Id,
},
})
}
func HandleHubMessage(response http.ResponseWriter, request *http.Request) {
@@ -292,10 +320,6 @@ func HandleHubMessage(response http.ResponseWriter, request *http.Request) {
return
}
channel.Mu.RLock()
perms := channel.UsersCachedPermissions
channel.Mu.RUnlock()
msg := &types.Message{
Id: uuid.New(),
AttachedFile: attachedFile,
@@ -306,20 +330,10 @@ func HandleHubMessage(response http.ResponseWriter, request *http.Request) {
}
channel.AddMessageToBuff(msg)
for permId, perm := range perms {
if !perm.CanReadHistory() {
continue
}
target, err := cache.GetUserById(permId)
if err != nil {
continue
}
wsServer.WsSendMessageCloseIfTimeout(target, &types.WsEventMessage{
Type: WsEventType.HubMessage,
Event: msg,
})
}
wsServer.WsSendEventMessageToPermittedChannelUsersCloseIfTimeout(user.Id, channel, types.CachedUserCanReadHistory, &types.WsEventMessage{
Type: WsEventType.HubMessage,
Event: msg,
})
response.WriteHeader(http.StatusCreated)
}
@@ -552,7 +566,7 @@ func hubPermissionContext(response http.ResponseWriter, request *http.Request, r
}
func HandleHubSetName(response http.ResponseWriter, request *http.Request) {
_, hub, ctx, _, ok := hubPermissionContext(response, request, normal, types.PermissionSetHubName)
requestor, hub, ctx, _, ok := hubPermissionContext(response, request, normal, types.PermissionSetHubName)
if !ok {
return
}
@@ -567,10 +581,14 @@ func HandleHubSetName(response http.ResponseWriter, request *http.Request) {
return
}
response.WriteHeader(http.StatusAccepted)
wsServer.WsSendEventMessageToHubUsersCloseIfTimeout(requestor.OriginalId, hub, &types.WsHubSpecificHubEventMessage{
Type: WsEventType.HubUpdate,
Event: &map[string]any{"name": newName},
})
}
func HandleHubSetColor(response http.ResponseWriter, request *http.Request) {
_, hub, ctx, _, ok := hubPermissionContext(response, request, normal, types.PermissionSetHubColor)
requestor, hub, ctx, _, ok := hubPermissionContext(response, request, normal, types.PermissionSetHubColor)
if !ok {
return
}
@@ -585,10 +603,14 @@ func HandleHubSetColor(response http.ResponseWriter, request *http.Request) {
return
}
response.WriteHeader(http.StatusAccepted)
wsServer.WsSendEventMessageToHubUsersCloseIfTimeout(requestor.OriginalId, hub, &types.WsHubSpecificHubEventMessage{
Type: WsEventType.HubUpdate,
Event: &map[string]any{"color": color},
})
}
func HandleHubRemove(response http.ResponseWriter, request *http.Request) {
_, hub, ctx, _, ok := hubPermissionContext(response, request, normal, types.PermissionRemoveHub)
requestor, hub, ctx, _, ok := hubPermissionContext(response, request, normal, types.PermissionRemoveHub)
if !ok {
return
}
@@ -596,12 +618,15 @@ func HandleHubRemove(response http.ResponseWriter, request *http.Request) {
http.Error(response, "db error", http.StatusInternalServerError)
return
}
wsServer.WsSendEventMessageToHubUsersCloseIfTimeout(requestor.OriginalId, hub, &types.WsHubSpecificHubEventMessage{
Type: WsEventType.HubRemove,
})
cache.DeleteHub(hub)
response.WriteHeader(http.StatusAccepted)
}
func HandleHubToggleUserColorAllowed(response http.ResponseWriter, request *http.Request) {
_, hub, ctx, _, ok := hubPermissionContext(response, request, normal, types.PermissionSetUserColorAllowed)
requestor, hub, ctx, _, ok := hubPermissionContext(response, request, normal, types.PermissionSetUserColorAllowed)
if !ok {
return
}
@@ -611,10 +636,14 @@ func HandleHubToggleUserColorAllowed(response http.ResponseWriter, request *http
return
}
response.WriteHeader(http.StatusAccepted)
wsServer.WsSendEventMessageToHubUsersCloseIfTimeout(requestor.OriginalId, hub, &types.WsHubSpecificHubEventMessage{
Type: WsEventType.HubUpdate,
Event: &map[string]any{"userColorAllowed": hub.UserColorAllowed},
})
}
func HandleSetHubJoinRole(response http.ResponseWriter, request *http.Request) {
_, hub, ctx, _, ok := hubPermissionContext(response, request, normal, types.PermissionSetHubJoinRole)
requestor, hub, ctx, _, ok := hubPermissionContext(response, request, normal, types.PermissionSetHubJoinRole)
if !ok {
return
}
@@ -634,10 +663,14 @@ func HandleSetHubJoinRole(response http.ResponseWriter, request *http.Request) {
return
}
response.WriteHeader(http.StatusAccepted)
wsServer.WsSendEventMessageToHubUsersCloseIfTimeout(requestor.OriginalId, hub, &types.WsHubSpecificHubEventMessage{
Type: WsEventType.HubUpdate,
Event: &map[string]any{"joinRole": newRole},
})
}
func HandleHubUserRemove(response http.ResponseWriter, request *http.Request) {
_, hub, ctx, usedRoleId, ok := hubPermissionContext(response, request, normal, types.PermissionRemoveUser)
requestor, hub, ctx, usedRoleId, ok := hubPermissionContext(response, request, normal, types.PermissionRemoveUser)
if !ok {
return
}
@@ -677,10 +710,19 @@ func HandleHubUserRemove(response http.ResponseWriter, request *http.Request) {
hub.Mu.Unlock()
updateChannelCacheForSpecUser(hubTarget, hub)
response.WriteHeader(http.StatusAccepted)
wsServer.WsSendEventMessageToHubUsersCloseIfTimeout(requestor.OriginalId, hub, &types.WsHubSpecificHubEventMessage{
Type: WsEventType.HubUserRemove,
Event: &map[string]any{"userId": targetId},
})
wsServer.WsSendMessageCloseIfTimeout(target, &types.WsHubSpecificHubEventMessage{
Type: WsEventType.HubUserRemove,
HubId: hub.Id,
Event: &map[string]any{"userId": targetId},
})
}
func HandleHubRenameUser(response http.ResponseWriter, request *http.Request) {
_, hub, ctx, usedRoleId, ok := hubPermissionContext(response, request, normal, types.PermissionRenameUser)
requestor, hub, ctx, usedRoleId, ok := hubPermissionContext(response, request, normal, types.PermissionRenameUser)
if !ok {
return
}
@@ -708,6 +750,10 @@ func HandleHubRenameUser(response http.ResponseWriter, request *http.Request) {
return
}
response.WriteHeader(http.StatusAccepted)
wsServer.WsSendEventMessageToHubUsersCloseIfTimeout(requestor.OriginalId, hub, &types.WsHubSpecificHubEventMessage{
Type: WsEventType.HubUserUpdate,
Event: target,
})
}
func HandleHubRenameSelf(response http.ResponseWriter, request *http.Request) {
@@ -721,10 +767,14 @@ func HandleHubRenameSelf(response http.ResponseWriter, request *http.Request) {
return
}
response.WriteHeader(http.StatusAccepted)
wsServer.WsSendEventMessageToHubUsersCloseIfTimeout(requestor.OriginalId, hub, &types.WsHubSpecificHubEventMessage{
Type: WsEventType.HubUserUpdate,
Event: requestor,
})
}
func HandleHubToggleMuteUser(response http.ResponseWriter, request *http.Request) {
_, hub, ctx, usedRoleId, ok := hubPermissionContext(response, request, normal, types.PermissionMuteUser)
requestor, hub, ctx, usedRoleId, ok := hubPermissionContext(response, request, normal, types.PermissionMuteUser)
if !ok {
return
}
@@ -751,10 +801,14 @@ func HandleHubToggleMuteUser(response http.ResponseWriter, request *http.Request
return
}
response.WriteHeader(http.StatusAccepted)
wsServer.WsSendEventMessageToHubUsersCloseIfTimeout(requestor.OriginalId, hub, &types.WsHubSpecificHubEventMessage{
Type: WsEventType.HubUserUpdate,
Event: target,
})
}
func HandleHubUserAddRole(response http.ResponseWriter, request *http.Request) {
_, hub, ctx, usedRoleId, ok := hubPermissionContext(response, request, normal, types.PermissionUserAddRole)
requestor, hub, ctx, usedRoleId, ok := hubPermissionContext(response, request, normal, types.PermissionUserAddRole)
if !ok {
return
}
@@ -792,10 +846,17 @@ func HandleHubUserAddRole(response http.ResponseWriter, request *http.Request) {
return
}
response.WriteHeader(http.StatusAccepted)
wsServer.WsSendEventMessageToHubUsersCloseIfTimeout(requestor.OriginalId, hub, &types.WsHubSpecificHubEventMessage{
Type: WsEventType.HubUserUpdate,
Event: target,
})
if targetUser, err := cache.GetUserById(targetId); err == nil {
sendChannelUpdatesToUser(targetUser, hub)
}
}
func HandleHubUserRemoveRole(response http.ResponseWriter, request *http.Request) {
_, hub, ctx, usedRoleId, ok := hubPermissionContext(response, request, normal, types.PermissionUserRemoveRole)
requestor, hub, ctx, usedRoleId, ok := hubPermissionContext(response, request, normal, types.PermissionUserRemoveRole)
if !ok {
return
}
@@ -828,10 +889,17 @@ func HandleHubUserRemoveRole(response http.ResponseWriter, request *http.Request
return
}
response.WriteHeader(http.StatusAccepted)
wsServer.WsSendEventMessageToHubUsersCloseIfTimeout(requestor.OriginalId, hub, &types.WsHubSpecificHubEventMessage{
Type: WsEventType.HubUserUpdate,
Event: target,
})
if targetUser, err := cache.GetUserById(targetId); err == nil {
sendChannelUpdatesToUser(targetUser, hub)
}
}
func HandleHubCreateRole(response http.ResponseWriter, request *http.Request) {
_, hub, ctx, usedRoleId, ok := hubPermissionContext(response, request, normal, types.PermissionCreateRole)
requestor, hub, ctx, usedRoleId, ok := hubPermissionContext(response, request, normal, types.PermissionCreateRole)
if !ok {
return
}
@@ -868,10 +936,14 @@ func HandleHubCreateRole(response http.ResponseWriter, request *http.Request) {
return
}
response.WriteHeader(http.StatusCreated)
wsServer.WsSendEventMessageToHubUsersCloseIfTimeout(requestor.OriginalId, hub, &types.WsHubSpecificHubEventMessage{
Type: WsEventType.HubRoleCreate,
Event: newRole,
})
}
func HandleHubRemoveRole(response http.ResponseWriter, request *http.Request) {
_, hub, ctx, usedRoleId, ok := hubPermissionContext(response, request, normal, types.PermissionRemoveRole)
requestor, hub, ctx, usedRoleId, ok := hubPermissionContext(response, request, normal, types.PermissionRemoveRole)
if !ok {
return
}
@@ -902,10 +974,14 @@ func HandleHubRemoveRole(response http.ResponseWriter, request *http.Request) {
return
}
response.WriteHeader(http.StatusAccepted)
wsServer.WsSendEventMessageToHubUsersCloseIfTimeout(requestor.OriginalId, hub, &types.WsHubSpecificHubEventMessage{
Type: WsEventType.HubRoleRemove,
Event: &map[string]any{"roleId": targetRoleId},
})
}
func HandleRoleSetName(response http.ResponseWriter, request *http.Request) {
_, hub, ctx, usedRoleId, ok := hubPermissionContext(response, request, normal, types.PermissionSetRoleName)
requestor, hub, ctx, usedRoleId, ok := hubPermissionContext(response, request, normal, types.PermissionSetRoleName)
if !ok {
return
}
@@ -938,10 +1014,14 @@ func HandleRoleSetName(response http.ResponseWriter, request *http.Request) {
return
}
response.WriteHeader(http.StatusAccepted)
wsServer.WsSendEventMessageToHubUsersCloseIfTimeout(requestor.OriginalId, hub, &types.WsHubSpecificHubEventMessage{
Type: WsEventType.HubRoleUpdate,
Event: role,
})
}
func HandleRoleSetColor(response http.ResponseWriter, request *http.Request) {
_, hub, ctx, usedRoleId, ok := hubPermissionContext(response, request, normal, types.PermissionSetRoleColor)
requestor, hub, ctx, usedRoleId, ok := hubPermissionContext(response, request, normal, types.PermissionSetRoleColor)
if !ok {
return
}
@@ -974,6 +1054,10 @@ func HandleRoleSetColor(response http.ResponseWriter, request *http.Request) {
return
}
response.WriteHeader(http.StatusAccepted)
wsServer.WsSendEventMessageToHubUsersCloseIfTimeout(requestor.OriginalId, hub, &types.WsHubSpecificHubEventMessage{
Type: WsEventType.HubRoleUpdate,
Event: role,
})
}
func HandleRoleSetPermissions(response http.ResponseWriter, request *http.Request) {
@@ -990,9 +1074,9 @@ func HandleRoleSetPermissions(response http.ResponseWriter, request *http.Reques
http.Error(response, "target role higher in hierarchy", http.StatusForbidden)
return
}
hub.Mu.Lock()
hub.Mu.RLock()
targetRole := hub.Roles[targetRoleId]
hub.Mu.Unlock()
hub.Mu.RUnlock()
if targetRole == nil {
http.Error(response, "no such role", http.StatusNotFound)
return
@@ -1039,6 +1123,23 @@ func HandleRoleSetPermissions(response http.ResponseWriter, request *http.Reques
return
}
response.WriteHeader(http.StatusAccepted)
wsServer.WsSendEventMessageToHubUsersCloseIfTimeout(requestor.OriginalId, hub, &types.WsHubSpecificHubEventMessage{
Type: WsEventType.HubRoleUpdate,
Event: targetRole,
})
hub.Mu.RLock()
affectedHubUsers := make([]*types.HubUser, 0)
for _, u := range hub.Users {
if u.OriginalId != requestor.OriginalId && u.Roles.ContainsRoleId(targetRole.Id) {
affectedHubUsers = append(affectedHubUsers, u)
}
}
hub.Mu.RUnlock()
for _, u := range affectedHubUsers {
if targetUser, err := cache.GetUserById(u.OriginalId); err == nil {
sendChannelUpdatesToUser(targetUser, hub)
}
}
}
func HandleRoleSelfRemove(response http.ResponseWriter, request *http.Request) {
@@ -1053,6 +1154,13 @@ func HandleRoleSelfRemove(response http.ResponseWriter, request *http.Request) {
return
}
response.WriteHeader(http.StatusAccepted)
wsServer.WsSendEventMessageToHubUsersCloseIfTimeout(requestor.OriginalId, hub, &types.WsHubSpecificHubEventMessage{
Type: WsEventType.HubUserUpdate,
Event: requestor,
})
if selfUser, err := cache.GetUserById(requestor.OriginalId); err == nil {
sendChannelUpdatesToUser(selfUser, hub)
}
}
func HandleChannelCreate(response http.ResponseWriter, request *http.Request) {
@@ -1099,10 +1207,15 @@ func HandleChannelCreate(response http.ResponseWriter, request *http.Request) {
return
}
response.WriteHeader(http.StatusAccepted)
wsServer.WsSendEventMessageToPermittedChannelUsersCloseIfTimeout(requestor.OriginalId, newHubChannel, types.CachedUserCanView, &types.WsHubSpecificHubEventMessage{
Type: WsEventType.HubChannelCreate,
HubId: hub.Id,
Event: newHubChannel,
})
}
func HandleChannelRemove(response http.ResponseWriter, request *http.Request) {
_, hub, ctx, _, ok := hubPermissionContext(response, request, normal, types.PermissionRemoveChannel)
requestor, hub, ctx, _, ok := hubPermissionContext(response, request, normal, types.PermissionRemoveChannel)
if !ok {
return
}
@@ -1112,12 +1225,20 @@ func HandleChannelRemove(response http.ResponseWriter, request *http.Request) {
return
}
hub.Mu.Lock()
if _, ok = hub.Channels[channelId]; !ok {
hub.Mu.Unlock()
var channel *types.HubChannel
hub.Mu.RLock()
channel, ok = hub.Channels[channelId]
hub.Mu.RUnlock()
if !ok {
http.Error(response, "no such channel", http.StatusNotFound)
return
}
wsServer.WsSendEventMessageToPermittedChannelUsersCloseIfTimeout(requestor.OriginalId, channel, types.CachedUserCanView, &types.WsHubSpecificHubEventMessage{
Type: WsEventType.HubChannelRemove,
HubId: hub.Id,
Event: &map[string]any{"channelId": channelId},
})
hub.Mu.Lock()
delete(hub.Channels, channelId)
hub.Mu.Unlock()
if err = postgresql.HubChannelDelete(ctx, channelId); err != nil {
@@ -1127,7 +1248,7 @@ func HandleChannelRemove(response http.ResponseWriter, request *http.Request) {
response.WriteHeader(http.StatusAccepted)
}
func HandleChannelSetName(response http.ResponseWriter, request *http.Request) {
_, hub, ctx, _, ok := hubPermissionContext(response, request, normal, types.PermissionSetChannelName)
requestor, hub, ctx, _, ok := hubPermissionContext(response, request, normal, types.PermissionSetChannelName)
if !ok {
return
}
@@ -1156,10 +1277,15 @@ func HandleChannelSetName(response http.ResponseWriter, request *http.Request) {
return
}
response.WriteHeader(http.StatusAccepted)
wsServer.WsSendEventMessageToPermittedChannelUsersCloseIfTimeout(requestor.OriginalId, channel, types.CachedUserCanView, &types.WsHubSpecificHubEventMessage{
Type: WsEventType.HubChannelUpdate,
HubId: hub.Id,
Event: &map[string]any{"channelId": channelId, "name": newName},
})
}
func HandleChannelSetDescription(response http.ResponseWriter, request *http.Request) {
_, hub, ctx, _, ok := hubPermissionContext(response, request, normal, types.PermissionSetChannelDescription)
requestor, hub, ctx, _, ok := hubPermissionContext(response, request, normal, types.PermissionSetChannelDescription)
if !ok {
return
}
@@ -1184,10 +1310,15 @@ func HandleChannelSetDescription(response http.ResponseWriter, request *http.Req
return
}
response.WriteHeader(http.StatusAccepted)
wsServer.WsSendEventMessageToPermittedChannelUsersCloseIfTimeout(requestor.OriginalId, channel, types.CachedUserCanView, &types.WsHubSpecificHubEventMessage{
Type: WsEventType.HubChannelUpdate,
HubId: hub.Id,
Event: &map[string]any{"channelId": channelId, "description": newDescription},
})
}
func handleChannelRolePermission(response http.ResponseWriter, request *http.Request, perm types.Permissions, updateList *types.HubChannelUpdate, modify func(*types.HubChannel, uint8, bool)) {
_, hub, ctx, usedRoleId, ok := hubPermissionContext(response, request, normal, perm)
requestor, hub, ctx, usedRoleId, ok := hubPermissionContext(response, request, normal, perm)
if !ok {
return
}
@@ -1217,12 +1348,33 @@ func handleChannelRolePermission(response http.ResponseWriter, request *http.Req
modify(channel, roleId, allow)
hub.Mu.Unlock()
channel.Mu.RLock()
viewers := make([]uuid.UUID, 0, len(channel.UsersCachedPermissions))
for id, perm := range channel.UsersCachedPermissions {
if id != requestor.OriginalId && perm.CanView() {
viewers = append(viewers, id)
}
}
channel.Mu.RUnlock()
updateChannelCacheForSpecChannel(channel, hub)
if err = postgresql.HubChannelUpdate(ctx, channel, updateList); err != nil {
http.Error(response, "db error", http.StatusInternalServerError)
return
}
response.WriteHeader(http.StatusAccepted)
msg := &types.WsHubSpecificHubEventMessage{
Type: WsEventType.HubChannelUpdate,
HubId: hub.Id,
Event: channel,
}
for _, id := range viewers {
target, err := cache.GetUserById(id)
if err != nil {
continue
}
wsServer.WsSendMessageCloseIfTimeout(target, msg)
}
}
func HandleChannelSetPermittedVisibleRole(response http.ResponseWriter, request *http.Request) {