1package webmail

2

3import (

4 "bufio"

5 "context"

6 "encoding/json"

7 "fmt"

8 "io"

9 "log/slog"

10 mathrand2 "math/rand/v2"

11 "net/http"

12 "runtime/debug"

13 "sync"

14 "time"

15

16 "github.com/mjl-/mox/metrics"

17 "github.com/mjl-/mox/mlog"

18 "github.com/mjl-/mox/store"

19)

20

21type eventWriter struct {

22 out writeFlusher

23 waitMin, waitMax time.Duration

24

25 // If connection is closed, the goroutine doing delayed writes must abort.

26 sync.Mutex

27 closed bool

28

29 // Before writing an event, we check if session is still valid. If not, we send a

30 // fatal error instead.

31 accountName string

32 sessionToken store.SessionToken

33

34 wrote bool // To be reset by user, set on write.

35 events chan struct {

36 name string // E.g. "start" for EventStart.

37 v any // Written as JSON.

38 when time.Time // For delaying.

39 } // Will only be set when waitMin or waitMax is > 0. Closed on connection shutdown.

40 errors chan error // If we have an events channel, we read errors and abort for them.

41}

42

43func newEventWriter(out writeFlusher, waitMin, waitMax time.Duration, accountName string, sessionToken store.SessionToken) *eventWriter {

44 return &eventWriter{out: out, waitMin: waitMin, waitMax: waitMax, accountName: accountName, sessionToken: sessionToken}

45}

46

47// close shuts down the events channel, causing the goroutine (if created) to

48// stop.

49func (ew *eventWriter) close() {

50 if ew.events != nil {

51 close(ew.events)

52 }

53 ew.Lock()

54 defer ew.Unlock()

55 ew.closed = true

56}

57

58// Write an event to the connection, e.g. "start" with value v, written as

59// JSON. This directly writes the event, no more delay.

60func (ew *eventWriter) write(name string, v any) error {

61 bw := bufio.NewWriter(ew.out)

62 if _, err := fmt.Fprintf(bw, "event: %s\ndata: ", name); err != nil {

63 return err

64 } else if err := json.NewEncoder(bw).Encode(v); err != nil {

65 return err

66 } else if _, err := fmt.Fprint(bw, "\n"); err != nil {

67 return err

68 } else if err := bw.Flush(); err != nil {

69 return err

70 }

71 return ew.out.Flush()

72}

73

74// Schedule an event for writing to the connection. If events get a delay, this

75// function still returns immediately.

76func (ew *eventWriter) xsendEvent(ctx context.Context, log mlog.Log, name string, v any) {

77 if name != "fatalErr" {

78 if _, err := store.SessionUse(ctx, log, ew.accountName, ew.sessionToken, ""); err != nil {

79 ew.xsendEvent(ctx, log, "fatalErr", "session no longer valid")

80 return

81 }

82 }

83

84 if (ew.waitMin > 0 || ew.waitMax > 0) && ew.events == nil {

85 // First write on a connection with delay.

86 ew.events = make(chan struct {

87 name string

88 v any

89 when time.Time

90 }, 100)

91 ew.errors = make(chan error)

92 go func() {

93 defer func() {

94 x := recover() // Should not happen, but don't take program down if it does.

95 if x != nil {

96 log.WithContext(ctx).Error("writeEvent panic", slog.Any("err", x))

97 debug.PrintStack()

98 metrics.PanicInc(metrics.Webmailsendevent)

99 }

100 }()

101

102 for {

103 ev, ok := <-ew.events

104 if !ok {

105 return

106 }

107 d := time.Until(ev.when)

108 if d > 0 {

109 time.Sleep(d)

110 }

111 ew.Lock()

112 if ew.closed {

113 ew.Unlock()

114 return

115 }

116 err := ew.write(ev.name, ev.v)

117 ew.Unlock()

118 if err != nil {

119 ew.errors <- err

120 return

121 }

122 }

123 }()

124 }

125 // Check for previous write error before continuing.

126 if ew.errors != nil {

127 select {

128 case err := <-ew.errors:

129 panic(ioErr{err})

130 default:

131 break

132 }

133 }

134 // If we have an events channel, we have a goroutine that write the events, delayed.

135 if ew.events != nil {

136 wait := ew.waitMin + time.Duration(mathrand2.IntN(1000))*(ew.waitMax-ew.waitMin)/1000

137 when := time.Now().Add(wait)

138 ew.events <- struct {

139 name string

140 v any

141 when time.Time

142 }{name, v, when}

143 } else {

144 err := ew.write(name, v)

145 if err != nil {

146 panic(ioErr{err})

147 }

148 }

149 ew.wrote = true

150}

151

152// writeFlusher is a writer and flusher. We need to flush after writing an

153// Event. Both to flush pending gzip data to the http response, and the http

154// response to the client.

155type writeFlusher interface {

156 io.Writer

157 Flush() error

158}

159

160// nopFlusher is a standin for writeFlusher if gzip is not used.

161type nopFlusher struct {

162 io.Writer

163}

164

165func (f nopFlusher) Flush() error {

166 return nil

167}

168

169// httpFlusher wraps Flush for a writeFlusher with a call to an http.Flusher.

170type httpFlusher struct {

171 writeFlusher

172 f http.Flusher

173}

174

175// Flush flushes the underlying writeFlusher, and calls Flush on the http.Flusher

176// (which doesn't return an error).

177func (f httpFlusher) Flush() error {

178 err := f.writeFlusher.Flush()

179 f.f.Flush()

180 return err

181}

182