Notification Dispatcher

1import java.util.*;
2import java.util.regex.*;
3/** Local dispatch coordinator. No real messages are sent by the included tests. */
4public final class Solution {
5    private Solution() {}
6    public enum Channel { EMAIL, SMS, PUSH }
7    public enum State { QUEUED, SENDING, ACCEPTED, DELIVERED, FAILED, CANCELLED, UNKNOWN }
8    public enum Outcome { ACCEPTED, RETRYABLE, PERMANENT, UNKNOWN }
9    public record Template(String version,String text){public Template{if(version==null||version.isBlank()||text==null)throw new IllegalArgumentException("template");}}
10    public record Delivery(String id,String recipient,Channel channel,String templateVersion,String body,State state,int attempts,long nextAt,String providerId){}
11    public record ProviderResult(Outcome outcome,String providerId){public ProviderResult{Objects.requireNonNull(outcome);if(providerId==null||(outcome==Outcome.ACCEPTED&&providerId.isBlank()))throw new IllegalArgumentException("provider ID");}}
12    @FunctionalInterface public interface Provider { ProviderResult send(Delivery delivery,String idempotencyKey) throws Exception; }
13    private record Request(String recipient,Channel channel,Template template,Map<String,String> fields){public Request{fields=Map.copyOf(fields);}}
14    private record Operation(Request request,String deliveryId){}
15    private record Receipt(String deliveryId,boolean delivered){}
16    public static String render(Template template,Map<String,String> values){
17        Objects.requireNonNull(template);var fields=Map.copyOf(values);var matcher=Pattern.compile("\\{([A-Za-z][A-Za-z0-9_]*)\\}").matcher(template.text());var output=new StringBuilder();var used=new HashSet<String>();
18        while(matcher.find()){String name=matcher.group(1),value=fields.get(name);if(value==null)throw new IllegalArgumentException("missing template field");used.add(name);matcher.appendReplacement(output,Matcher.quoteReplacement(value));}matcher.appendTail(output);
19        if(!used.equals(fields.keySet()))throw new IllegalArgumentException("unused fields");return output.toString();
20    }
21    public static final class Dispatcher {
22        private final Map<Channel,Provider> providers;
23        private final int maxAttempts;
24        private final long baseDelay;
25        private final Map<String,Set<Channel>> preferences=new HashMap<>();
26        private final Map<String,Delivery> deliveries=new LinkedHashMap<>();
27        private final Map<String,Operation> operations=new HashMap<>();
28        private final Map<String,Receipt> receipts=new HashMap<>();
29        private long sequence,time;
30        public Dispatcher(Map<Channel,Provider> providers,int maxAttempts,long baseDelay){this.providers=Map.copyOf(providers);if(maxAttempts<1||maxAttempts>10||baseDelay<1||baseDelay>Long.MAX_VALUE/1024)throw new IllegalArgumentException("retry bounds");this.maxAttempts=maxAttempts;this.baseDelay=baseDelay;}
31        private void at(long now){if(now<time)throw new IllegalArgumentException("time");time=now;}
32        public synchronized void setAllowed(String recipient,Set<Channel> channels){if(recipient==null||recipient.isBlank())throw new IllegalArgumentException("recipient");preferences.put(recipient,Set.copyOf(channels));}
33        public synchronized Delivery submit(String key,String recipient,Channel channel,Template template,Map<String,String> fields){
34            if(key==null||key.isBlank()||recipient==null||recipient.isBlank()||!providers.containsKey(channel))throw new IllegalArgumentException("request");
35            var request=new Request(recipient,channel,template,fields);var old=operations.get(key);if(old!=null){if(!old.request().equals(request))throw new IllegalArgumentException("key reused");return get(old.deliveryId());}
36            String body=render(template,fields);var delivery=new Delivery("D"+(++sequence),recipient,channel,template.version(),body,State.QUEUED,0,0,"");deliveries.put(delivery.id(),delivery);operations.put(key,new Operation(request,delivery.id()));return delivery;
37        }
38        private static Delivery change(Delivery d,State state,int attempts,long next,String providerId){return new Delivery(d.id(),d.recipient(),d.channel(),d.templateVersion(),d.body(),state,attempts,next,providerId);}
39        public Delivery dispatch(String id,long now){
40            final Delivery claimed;final long retryTime;
41            synchronized(this){
42                at(now);var d=get(id);if(d.state()!=State.QUEUED||now<d.nextAt())throw new IllegalStateException("not due");
43                if(!preferences.getOrDefault(d.recipient(),Set.of()).contains(d.channel())){var cancelled=change(d,State.CANCELLED,d.attempts(),d.nextAt(),d.providerId());deliveries.put(id,cancelled);return cancelled;}
44                retryTime=Math.addExact(now,Math.multiplyExact(baseDelay,1L<<d.attempts()));
45                claimed=change(d,State.SENDING,d.attempts()+1,d.nextAt(),d.providerId());deliveries.put(id,claimed);
46            }
47            ProviderResult result;
48            try{result=Objects.requireNonNull(providers.get(claimed.channel()).send(claimed,claimed.id()));}
49            catch(Exception e){if(e instanceof InterruptedException)Thread.currentThread().interrupt();result=new ProviderResult(Outcome.UNKNOWN,"");}
50            synchronized(this){
51                var current=get(id);
52                // A delivery webhook may have completed while the request was in flight.
53                if(current.state()!=State.SENDING)return current;
54                State state=switch(result.outcome()){case ACCEPTED->State.ACCEPTED;case RETRYABLE->claimed.attempts()<maxAttempts?State.QUEUED:State.FAILED;case PERMANENT->State.FAILED;case UNKNOWN->State.UNKNOWN;};
55                var updated=change(claimed,state,claimed.attempts(),state==State.QUEUED?retryTime:claimed.nextAt(),result.providerId());deliveries.put(id,updated);return updated;
56            }
57        }
58        public synchronized Delivery receipt(String eventId,String deliveryId,boolean delivered){
59            if(eventId==null||eventId.isBlank())throw new IllegalArgumentException("event");var old=receipts.get(eventId);var receipt=new Receipt(deliveryId,delivered);
60            if(old!=null){if(!old.equals(receipt))throw new IllegalArgumentException("event reused");return get(deliveryId);}
61            var d=get(deliveryId);if(!Set.of(State.SENDING,State.ACCEPTED,State.UNKNOWN,State.DELIVERED,State.FAILED).contains(d.state()))throw new IllegalStateException("no provider attempt");
62            if(d.attempts()==0)throw new IllegalStateException("no provider attempt");
63            receipts.put(eventId,receipt);
64            if(d.state()==State.DELIVERED)return d; // A late failure cannot regress delivered.
65            var next=change(d,delivered?State.DELIVERED:State.FAILED,d.attempts(),d.nextAt(),d.providerId());deliveries.put(deliveryId,next);return next;
66        }
67        public synchronized Delivery cancel(String id){var d=get(id);if(d.state()!=State.QUEUED&&d.state()!=State.CANCELLED)throw new IllegalStateException("cannot retract provider attempt");var next=change(d,State.CANCELLED,d.attempts(),d.nextAt(),d.providerId());deliveries.put(id,next);return next;}
68        public synchronized Delivery get(String id){var d=deliveries.get(id);if(d==null)throw new IllegalArgumentException("delivery");return d;}
69    }
70}
71