package async_smtp

  1. Overview
  2. Docs
Legend:
Page
Library
Module
Module type
Parameter
Class
Class type
Source

Source file spool.ml

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
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
open Core
open Async
open! Async_smtp
open Common
module Time = Time_float_unix

let msgid = Command.Param.Arg_type.create Smtp_spool.Stable.Message_id.V1.of_string

module Client_side_filter = struct
  type t =
    { next_hop : Re2.t option
    ; sender : Re2.t option
    ; recipient : Re2.t option
    ; queue : Smtp_spool_message.Queue.t option
    ; younger_than : Time_float_unix.Span.t option
    ; older_than : Time_float_unix.Span.t option
    }
  [@@deriving fields ~iterators:fold]

  let has_some t =
    let or_is_some acc field = acc || Option.is_some (Field.get field t) in
    Fields.fold
      ~init:false
      ~next_hop:or_is_some
      ~sender:or_is_some
      ~recipient:or_is_some
      ~queue:or_is_some
      ~younger_than:or_is_some
      ~older_than:or_is_some
  ;;

  let param_opt =
    let open Command.Let_syntax in
    let open Command.Param in
    let re_opt_param ~name =
      flag
        ~doc:[%string "REGEX filter %{name} by regex"]
        name
        (Arg_type.create Re2.of_string |> optional)
    in
    let than_opt_param ~name =
      flag ~doc:[%string "TIME %{name} than"] name (optional Time.Span.arg_type)
    in
    let%map_open next_hop = re_opt_param ~name:"next-hop"
    and sender = re_opt_param ~name:"sender"
    and recipient = re_opt_param ~name:"recipient"
    and queue =
      flag
        ~doc:"QUEUE the status/queue of the message"
        ~aliases:[ "status" ]
        "queue"
        (Arg_type.enumerated (module Smtp_spool_message.Queue) |> optional)
    and younger_than = than_opt_param ~name:"younger"
    and older_than = than_opt_param ~name:"older" in
    (* If all fields are [None], return [None]. Only return [Some t] if [Some] field has a
       value. *)
    let t = { next_hop; sender; recipient; queue; younger_than; older_than } in
    Option.some_if (has_some t) t
  ;;

  let matches ~now t msg =
    let module S = Smtp_spool.Spooled_message_info in
    let or_matches m field ~f =
      match m with
      | true -> true
      | false as default -> Option.value_map ~default (Field.get field t) ~f
    in
    let or_matches_re m field ~get_values ~value_to_string =
      or_matches m field ~f:(fun re ->
        get_values msg |> List.map ~f:value_to_string |> List.exists ~f:(Re2.matches re))
    in
    Fields.fold
      ~init:false
      ~next_hop:
        (or_matches_re
           ~get_values:S.next_hop_choices
           ~value_to_string:Host_and_port.to_string)
      ~sender:
        (or_matches_re
           ~get_values:(fun msg -> [ S.envelope_info msg |> Smtp_envelope.Info.sender ])
           ~value_to_string:Smtp_envelope.Sender.to_string)
      ~recipient:
        (or_matches_re
           ~get_values:(fun msg -> S.envelope_info msg |> Smtp_envelope.Info.recipients)
           ~value_to_string:Email_address.to_string)
      ~queue:
        (or_matches ~f:(fun queue ->
           S.status msg
           |> Smtp_spool_message.Queue.of_status
           |> Option.value_map
                ~default:false
                ~f:([%compare.equal: Smtp_spool_message.Queue.t] queue)))
      ~younger_than:
        (or_matches ~f:(fun younger_than ->
           Time.Span.(Time.diff now (S.spool_date msg) < younger_than)))
      ~older_than:
        (or_matches ~f:(fun older_than ->
           Time.Span.(Time.diff now (S.spool_date msg) > older_than)))
  ;;

  let filter ?now t msgs =
    let now = Option.value_or_thunk now ~default:Time.now in
    List.filter msgs ~f:(matches ~now t)
  ;;

  let filter_opt ?now t msgs =
    match t with
    | None -> msgs
    | Some t -> filter ?now t msgs
  ;;
end

module Status = struct
  module Format = struct
    type t =
      [ `Ascii_table
      | `Ascii_table_with_max_width of int
      | `Exim
      | `Sexp
      | `Id
      ]
    [@@deriving sexp]

    let of_string = function
      | "ascii" -> `Ascii_table
      | "exim" -> `Exim
      | "sexp" -> `Sexp
      | "id" -> `Id
      | str -> Sexp.of_string_conv_exn str [%of_sexp: t]
    ;;

    let arg_type = Command.Param.Arg_type.create of_string

    let param =
      Command.Param.(
        flag
          "format"
          (optional_with_default `Ascii_table arg_type)
          ~doc:
            " Output format for the spool, valid values include 'ascii', 'exim', 'sexp', \
             'id'")
    ;;
  end

  let dispatch ~format ?client_side_filter client =
    let%map status =
      Rpc.Rpc.dispatch_exn Smtp_rpc_intf.Spool.status client ()
      >>| Client_side_filter.filter_opt client_side_filter
    in
    printf "%s\n" (Smtp_spool.Status.to_formatted_string ~format status)
  ;;

  let on_disk ~format ?client_side_filter config =
    let%map status =
      Smtp_spool.status_from_disk config
      |> Deferred.Or_error.ok_exn
      >>| Client_side_filter.filter_opt client_side_filter
    in
    printf "%s\n" (Smtp_spool.Status.to_formatted_string ~format status)
  ;;

  let command =
    let open Command.Let_syntax in
    Command.configs_or_rpc
      ~summary:"list current contents of the spool"
      [%map_open
        let format = Format.param
        and client_side_filter = Client_side_filter.param_opt in
        function
        | `Configs (_, spool_config) -> on_disk ~format ?client_side_filter spool_config
        | `Rpc client -> dispatch ~format ?client_side_filter client]
  ;;
end

module Count = struct
  let which =
    Command.Param.(
      choose_one
        ~if_nothing_chosen:(Default_to `All)
        [ flag "-frozen-only" no_arg ~doc:" Only count frozen messages"
          |> map ~f:(fun b -> Option.some_if b `Only_frozen)
        ; flag "-active-only" no_arg ~doc:" Only count active messages"
          |> map ~f:(fun b -> Option.some_if b `Only_active)
        ])
  ;;

  let is_frozen = function
    | `Frozen -> true
    | `Send_now | `Send_at _ | `Sending | `Removed | `Delivered | `Quarantined _ -> false
  ;;

  let is_active = function
    | `Send_now | `Send_at _ | `Sending -> true
    | `Frozen | `Removed | `Delivered | `Quarantined _ -> false
  ;;

  let dispatch ~which ?client_side_filter client =
    Rpc.Rpc.dispatch_exn Smtp_rpc_intf.Spool.status client ()
    >>| Client_side_filter.filter_opt client_side_filter
    >>| List.filter ~f:(fun message_info ->
          let status = Smtp_spool.Spooled_message_info.status message_info in
          match which with
          | `All -> true
          | `Only_frozen -> is_frozen status
          | `Only_active -> is_active status)
    >>| (List.length :> _ -> _)
  ;;

  let on_disk ?client_side_filter config =
    let%map count =
      match client_side_filter with
      | None -> Smtp_spool.count_from_disk config >>| Or_error.ok_exn
      | Some client_side_filter ->
        let%map l =
          Smtp_spool.status_from_disk config
          |> Deferred.Or_error.ok_exn
          >>| Client_side_filter.filter client_side_filter
        in
        List.length l
    in
    printf "%d\n" count
  ;;

  let command =
    let open Command.Let_syntax in
    Command.configs_or_rpc
      ~summary:"print total number of messages in the spool"
      [%map_open
        let which = which
        and client_side_filter = Client_side_filter.param_opt in
        function
        | `Configs (_, spool_config) -> on_disk ?client_side_filter spool_config
        | `Rpc client ->
          let open Deferred.Let_syntax in
          dispatch ~which ?client_side_filter client >>| printf "%d\n"]
  ;;
end

module Freeze = struct
  let msgids = Command.Param.(anon (sequence ("msgid" %: msgid)))

  let dispatch ~msgids client =
    Rpc.Rpc.dispatch_exn Smtp_rpc_intf.Spool.freeze client msgids >>| Or_error.ok_exn
  ;;

  let command =
    let open Command.Let_syntax in
    Command.rpc
      ~summary:"Freeze messages in the spool"
      [%map_open
        let msgids = msgids in
        dispatch ~msgids]
  ;;
end

module Send = struct
  let retry_intervals =
    Command.Param.(
      flag
        "retry-interval"
        (listed Time.Span.arg_type)
        ~doc:"SPAN additional retry intervals (order matters)"
      |> map ~f:(List.map ~f:Smtp_envelope.Retry_interval.create))
  ;;

  let param =
    let open Command.Let_syntax in
    [%map_open
      let all = flag "all" no_arg ~doc:" force immediate sending of all messages"
      and frozen =
        flag "frozen" no_arg ~doc:" force immidiate resending of frozen messages only"
      and msgids = anon (sequence ("msgid" %: msgid)) in
      match msgids with
      | [] when frozen -> `Frozen_only
      | [] when all -> `All_messages
      | [] -> failwith "Must specify either msgids or -all or -frozen"
      | _ when all || frozen -> failwith "Can't specify msgids and -all or -frozen"
      | msgids -> `Some_messages msgids]
  ;;

  let dispatch ?(retry_intervals = []) send_info client =
    Rpc.Rpc.dispatch_exn Smtp_rpc_intf.Spool.send client (retry_intervals, send_info)
    >>| Or_error.ok_exn
  ;;

  let command =
    let open Command.Let_syntax in
    Command.rpc
      ~summary:"Force immediate sending of messages"
      [%map_open
        let retry_intervals = retry_intervals
        and send_info = param in
        dispatch ~retry_intervals send_info]
  ;;
end

module Remove = struct
  let msgids = Command.Param.(anon (sequence ("msgid" %: msgid)))

  let dispatch ~msgids client =
    Rpc.Rpc.dispatch_exn Smtp_rpc_intf.Spool.remove client msgids >>| Or_error.ok_exn
  ;;

  let command =
    let open Command.Let_syntax in
    Command.rpc
      ~summary:"Remove messages in the spool"
      [%map_open
        let msgids = msgids in
        dispatch ~msgids]
  ;;
end

module Recover = struct
  let param =
    let open Command.Let_syntax in
    [%map_open
      let msgs = anon (sequence ("msg" %: msgid))
      and quarantine = flag "from-quarantine" no_arg ~doc:" recover quarantined messages"
      and remove = flag "from-remove" no_arg ~doc:" recover removed messages" in
      match quarantine, remove with
      | true, false -> { Smtp_spool.Recover_info.msgs; from = `Quarantined }
      | false, true -> { Smtp_spool.Recover_info.msgs; from = `Removed }
      | _ -> failwith "Must specify exactly one of -from-quarantine or -from-remove"]
  ;;

  let dispatch recover_info client =
    Rpc.Rpc.dispatch_exn Smtp_rpc_intf.Spool.recover client recover_info
    >>| Or_error.ok_exn
  ;;

  let command =
    let open Command.Let_syntax in
    Command.rpc
      ~summary:"recover a removed message back into a frozen state"
      [%map_open
        let recover_info = param in
        dispatch recover_info]
  ;;
end

module Events = struct
  let dispatch client =
    let%bind pipe, _ = Rpc.Pipe_rpc.dispatch_exn Smtp_rpc_intf.Spool.events client () in
    Pipe.iter_without_pushback pipe ~f:(fun event ->
      printf !"%{sexp:Smtp_spool.Event.t}\n" event)
  ;;

  let command =
    Command.rpc ~summary:"view the stream of spool events" (Command.Param.return dispatch)
  ;;
end

let command =
  Command.group
    ~summary:"spool management"
    [ "status", Status.command
    ; "count", Count.command
    ; "freeze", Freeze.command
    ; "send", Send.command
    ; "remove", Remove.command
    ; "recover", Recover.command
    ; "events", Events.command
    ]
;;
OCaml

Innovation. Community. Security.