Repository navigation
Expand file tree
/
Copy pathwebhooks.go
More file actions
executable file
·222 lines (180 loc) · 6.41 KB
/
Copy pathwebhooks.go
File metadata and controls
executable file
·222 lines (180 loc) · 6.41 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
package app
import (
"bytes"
"context"
"encoding/json"
"fmt"
"net/http"
"time"
log "github.com/golang/glog"
"nuntius/channel"
"nuntius/subscription"
"nuntius/utils"
)
const maxTimeout = 3 * time.Second
// A webHook is sent as a HTTP POST request to the url which you specify.
// The POST request payload (body) contains a JSON document, and follows the following format:
// {
// "time_ms": 1327078148132
// "events": [
// { "name": "event_name", "some": "data" }
// ]
// }
//
// Security
// Encryption
//
// You may use a HTTP or a HTTPS url for WebHooks. In most cases HTTP is sufficient, but HTTPS can be useful if your data is sensitive or if you wish to protect against replay attacks for example.
// Authentication
//
// Since anyone could in principle send WebHooks to your application, it’s important to verify that these WebHooks originated from Pusher. Valid WebHooks will therefore contain these headers which contain a HMAC signature of the webHook payload (body):
//
// X-Pusher-Key: The App Key.
// X-Pusher-Signature: A HMAC SHA256 hex digest formed by signing the POST payload (body) with the token’s secret.
type webHook struct {
TimeMs int64 `json:"time_ms"`
Events []hookEvent `json:"events"`
}
type hookEvent struct {
Name string `json:"name"`
Channel string `json:"channel"`
Event string `json:"event,omitempty"`
Data interface{} `json:"data,omitempty"`
SocketID string `json:"socket_id,omitempty"`
UserID string `json:"user_id,omitempty"`
}
func newChannelOcuppiedHook(channel *channel.Channel) hookEvent {
return hookEvent{Name: "channel_occupied", Channel: channel.ID}
}
func newChannelVacatedHook(channel *channel.Channel) hookEvent {
return hookEvent{Name: "channel_vacated", Channel: channel.ID}
}
func newMemberAddedHook(channel *channel.Channel, s *subscription.Subscription) hookEvent {
return hookEvent{Name: "member_added", Channel: channel.ID, UserID: s.ID}
}
func newMemberRemovedHook(channel *channel.Channel, s *subscription.Subscription) hookEvent {
return hookEvent{Name: "member_removed", Channel: channel.ID, UserID: s.ID}
}
func newClientHook(channel *channel.Channel, s *subscription.Subscription, event string, data interface{}) hookEvent {
return hookEvent{Name: "client_event", Channel: channel.ID, Event: event, Data: data, SocketID: s.Connection.SocketID}
}
// TriggerChannelOccupiedHook channel_occupied
// { "name": "channel_occupied", "channel": "test_channel" }
func (a *Application) TriggerChannelOccupiedHook(c *channel.Channel) {
event := newChannelOcuppiedHook(c)
ctx, cancel := context.WithTimeout(context.Background(), maxTimeout)
defer cancel()
if err := triggerHook(ctx, a, event); err != nil {
log.Errorf("triggering webhook %+v", err)
}
}
// TriggerChannelVacatedHook channel_vacated
// { "name": "channel_vacated", "channel": "test_channel" }
func (a *Application) TriggerChannelVacatedHook(c *channel.Channel) {
event := newChannelVacatedHook(c)
ctx, cancel := context.WithTimeout(context.Background(), maxTimeout)
defer cancel()
if err := triggerHook(ctx, a, event); err != nil {
log.Errorf("triggering webhook %+v", err)
}
}
// TriggerClientEventHook client_events
// {
// "name": "client_event",
// "channel": "name of the channel the event was published on",
// "event": "name of the event",
// "data": "data associated with the event",
// "socket_id": "socket_id of the sending socket",
// "user_id": "user_id associated with the sending socket" # Only for presence channels
// }
func (a *Application) TriggerClientEventHook(c *channel.Channel, s *subscription.Subscription, clientEvent string, data interface{}) {
event := newClientHook(c, s, clientEvent, data)
if c.IsPresence() {
event.UserID = s.ID
}
ctx, cancel := context.WithTimeout(context.Background(), maxTimeout)
defer cancel()
if err := triggerHook(ctx, a, event); err != nil {
log.Errorf("triggering webhook %+v", err)
}
}
// TriggerMemberAddedHook member_added
// {
// "name": "member_added",
// "channel": "presence-your_channel_name",
// "user_id": "a_user_id"
// }
func (a *Application) TriggerMemberAddedHook(c *channel.Channel, s *subscription.Subscription) {
event := newMemberAddedHook(c, s)
ctx, cancel := context.WithTimeout(context.Background(), maxTimeout)
defer cancel()
if err := triggerHook(ctx, a, event); err != nil {
log.Errorf("triggering webhook %+v", err)
}
}
// TriggerMemberRemovedHook member_removed
// {
// "name": "member_removed",
// "channel": "presence-your_channel_name",
// "user_id": "a_user_id"
// }
func (a *Application) TriggerMemberRemovedHook(c *channel.Channel, s *subscription.Subscription) {
event := newMemberRemovedHook(c, s)
ctx, cancel := context.WithTimeout(context.Background(), maxTimeout)
defer cancel()
if err := triggerHook(ctx, a, event); err != nil {
log.Errorf("triggering webhook %+v", err)
}
}
func triggerHook(ctx context.Context, a *Application, event hookEvent) error {
if !a.WebHooks {
log.Infof("webhook are not enabled for app: %s", a.Name)
return fmt.Errorf("webhooks are not enabled for app: %s", a.Name)
}
done := make(chan bool)
go func() {
log.Infof("Triggering %s event", event.Name)
hook := webHook{TimeMs: time.Now().Unix()}
hook.Events = append(hook.Events, event)
var js []byte
var err error
js, err = json.Marshal(hook)
if err != nil {
log.Errorf("Error decoding json: %+v", err)
return
}
var req *http.Request
req, err = http.NewRequest("POST", a.URLWebHook, bytes.NewReader(js))
if err != nil {
log.Errorf("Error creating request: %+v", err)
return
}
req = req.WithContext(ctx)
req.Header.Set("User-Agent", "Ipe UA; (+https://github.com/dimiro1/ipe)")
req.Header.Set("Content-Type", "application/json")
req.Header.Set("X-Pusher-Key", a.Key)
req.Header.Set("X-Pusher-Signature", utils.HashMAC(js, []byte(a.Secret)))
log.V(1).Infof("%+v", req.Header)
log.V(1).Infof("%+v", string(js))
resp, err := http.DefaultClient.Do(req)
// See: http://devs.cloudimmunity.com/gotchas-and-common-mistakes-in-go-golang/index.html#close_http_resp_body
if resp != nil {
defer func() {
if err := resp.Body.Close(); err != nil {
log.Errorf("error closing response body %+v", err)
}
}()
}
if err != nil {
log.Errorf("error posting %s event: %+v", event.Name, err)
}
// Successfully terminated
done <- true
}()
select {
case <-ctx.Done():
return ctx.Err()
case <-done:
return nil
}
}