Design an analytics library.

package com.androidinterview.analytics;

import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.LongSupplier;

// The only thing a call site sees. track() validates, stamps identity, writes
// to disk and returns. No network call, no lock held on anything slow, and
// nothing that can throw in a release build.
public final class Analytics {

    // Asking WorkManager for a one time flush. Keeping it a seam is what lets
    // this class be tested without Android, and it is also where the unique
    // work name goes, so two requests never run two workers over one queue.
    public interface FlushScheduler {
        void requestFlush();
    }

    // The schema, event name to the properties that event must carry. This is
    // the thing a logging library has no reason to own. A typo in an event
    // name is invisible until someone builds a dashboard on it a month later,
    // so the library refuses the event at the call site instead.
    private final Map<String, Set<String>> schema;
    private final List<EventQueue> destinations;
    private final Identity identity;
    private final FlushScheduler scheduler;
    private final LongSupplier now;
    // Debug builds throw so a bad event fails in review. Release builds drop
    // and count, because an analytics library must never crash the feature it
    // is measuring.
    private final boolean strict;

    private volatile boolean enabled = true;
    private volatile int rejected;

    public Analytics(
            Map<String, Set<String>> schema,
            List<EventQueue> destinations,
            Identity identity,
            FlushScheduler scheduler,
            LongSupplier now,
            boolean strict) {
        this.schema = Map.copyOf(schema);
        this.destinations = List.copyOf(destinations);
        this.identity = identity;
        this.scheduler = scheduler;
        this.now = now;
        this.strict = strict;
    }

    // The consent gate sits in front of the queue, not in front of the
    // uploader. Opting out drops what is already on disk, because those events
    // were collected under a consent that no longer exists.
    public void setTrackingEnabled(boolean value) {
        enabled = value;
        if (!value) {
            destinations.forEach(EventQueue::clear);
        }
    }

    public void track(String name, Map<String, Object> properties) {
        if (!enabled) {
            return;
        }
        if (!isValid(name, properties)) {
            rejected++;
            if (strict) {
                throw new IllegalArgumentException("Event does not match the schema, " + name);
            }
            return;
        }
        deliver(AnalyticsEvent.of(name, properties, identity, now.getAsLong()));
    }

    public void login(String userId) {
        // The alias goes to the queue even when the user has opted out of
        // product analytics on some setups, so this deliberately runs the same
        // gate as everything else and nothing special.
        if (enabled) {
            deliver(identity.login(userId, now.getAsLong()));
        }
    }

    public void logout() {
        identity.logout();
    }

    // One of the three flush triggers. The app going to the background is the
    // most valuable one, because it is the last moment before the process can
    // be killed, and the events from the session that just ended are sitting
    // in the queue.
    public void onAppBackgrounded() {
        scheduler.requestFlush();
    }

    // Fan out. The call site names the event once and every configured
    // destination gets its own copy in its own queue, so a third party backend
    // being down cannot hold up the first party one.
    private void deliver(AnalyticsEvent event) {
        boolean full = false;
        for (EventQueue queue : destinations) {
            full |= queue.append(event);
        }
        if (full) {
            scheduler.requestFlush();
        }
    }

    private boolean isValid(String name, Map<String, Object> properties) {
        Set<String> required = schema.get(name);
        return required != null && properties.keySet().containsAll(required);
    }

    public int rejectedCount() {
        return rejected;
    }
}

package com.androidinterview.analytics;

import java.util.Map;
import java.util.UUID;

// One tracked event. The id is minted on the device inside track() and never
// changes, not even across a retry, and that is the whole idempotency story.
// A batch the server accepted whose acknowledgement was lost gets sent again,
// and every event in it arrives carrying the id it already had, so the backend
// drops the duplicate instead of double counting the funnel.
public record AnalyticsEvent(
        String id,
        String name,
        Map<String, Object> properties,
        // The device clock at track() time. The server stamps its own receive
        // time beside it, because a phone with a wrong clock is common and the
        // backend needs both numbers to notice.
        long timestampMillis,
        String anonymousId,
        // Null until the user logs in. Everything before that is stitched by
        // the alias event Identity emits.
        String userId,
        String sessionId) {

    public AnalyticsEvent {
        properties = Map.copyOf(properties);
    }

    static AnalyticsEvent of(
            String name, Map<String, Object> properties, Identity identity, long nowMillis) {
        return new AnalyticsEvent(
                UUID.randomUUID().toString(),
                name,
                properties,
                nowMillis,
                identity.anonymousId(),
                identity.userId(),
                identity.sessionId(nowMillis));
    }
}

package com.androidinterview.analytics;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.UUID;
import java.util.stream.Collectors;

// What the background worker calls. Everything here runs off the calling
// thread, so a track() on the main thread never waits on a socket.
public final class BatchUploader {

    // Three outcomes, not two. Accepted and a transient failure are obvious,
    // and the third one is the one people forget. A batch the server rejects
    // outright, a schema the backend no longer accepts, retries forever and
    // blocks every event behind it, so it is dropped on purpose.
    public enum Outcome {
        ACCEPTED,
        TRANSIENT_FAILURE,
        REJECTED
    }

    public interface Transport {
        Outcome send(String batchId, List<AnalyticsEvent> events) throws IOException;
    }

    private final EventQueue queue;
    private final Transport transport;

    public BatchUploader(EventQueue queue, Transport transport) {
        this.queue = queue;
        this.transport = transport;
    }

    // The batch id is derived from the ids of the events in the batch rather
    // than minted per attempt. A retry of the same events therefore carries
    // the same batch id, so the server can recognise the replay in one lookup
    // instead of deduplicating event by event.
    static String batchId(List<AnalyticsEvent> events) {
        String joined = events.stream().map(AnalyticsEvent::id).collect(Collectors.joining(","));
        return UUID.nameUUIDFromBytes(joined.getBytes(StandardCharsets.UTF_8)).toString();
    }

    // Returns true when the queue was drained. False means a transient
    // failure, and the caller turns that into WorkManager's retry, which owns
    // the backoff. Writing a second backoff timer in here would just fight it.
    public boolean flush() {
        while (true) {
            List<AnalyticsEvent> batch = queue.nextBatch();
            if (batch.isEmpty()) {
                return true;
            }
            Outcome outcome;
            try {
                outcome = transport.send(batchId(batch), batch);
            } catch (IOException networkFailure) {
                return false;
            }
            if (outcome == Outcome.TRANSIENT_FAILURE) {
                return false;
            }
            // A rejected batch is acknowledged too, which reads wrong for a
            // second and is right. Leaving it queued means a poison batch that
            // nothing behind it can get past.
            queue.acknowledge(batch);
        }
    }
}

package com.androidinterview.analytics;

import java.util.List;

// The queue for one destination. It is persisted, because an in memory list
// loses every event tracked since the last flush the moment the process dies,
// and the screens users bounce off fastest are the ones that die first.
//
// One queue per destination, not one queue with a destination column. A
// backend that is down then backs up only its own queue, and the healthy
// destination keeps draining.
public final class EventQueue {

    // Room, or an append only file. The queue does not care which, it only
    // needs the rows back in the order they went in.
    public interface Storage {
        void append(AnalyticsEvent event);

        List<AnalyticsEvent> oldest(int limit);

        void delete(List<String> eventIds);

        int size();
    }

    private final String destination;
    private final Storage storage;
    private final int batchSize;
    private final int capacity;

    private int dropped;

    public EventQueue(String destination, Storage storage, int batchSize, int capacity) {
        this.destination = destination;
        this.storage = storage;
        this.batchSize = batchSize;
        this.capacity = capacity;
    }

    public String destination() {
        return destination;
    }

    // Returns true when the count trigger has fired, which is one of the three
    // reasons a flush happens. The other two are the periodic worker and the
    // app going to the background, and both of those live outside this class
    // because they are lifecycle, not queueing.
    //
    // The cap is the important line here. A user offline for a week must not
    // come back to fifty thousand queued events and one flush that hammers the
    // radio, so the oldest event is dropped to make room and the drop is
    // counted. Reporting the count matters more than it looks, because a
    // silent drop shows up later as a funnel that quietly lost its top.
    public synchronized boolean append(AnalyticsEvent event) {
        if (storage.size() >= capacity) {
            List<AnalyticsEvent> victims = storage.oldest(1);
            if (!victims.isEmpty()) {
                storage.delete(List.of(victims.get(0).id()));
                dropped++;
            }
        }
        storage.append(event);
        return storage.size() >= batchSize;
    }

    public synchronized List<AnalyticsEvent> nextBatch() {
        return storage.oldest(batchSize);
    }

    // Deleting happens only after the server has acknowledged the batch. The
    // cost of that order is a duplicate when the acknowledgement is lost, and
    // the event id is what makes that duplicate free to throw away.
    public synchronized void acknowledge(List<AnalyticsEvent> batch) {
        storage.delete(batch.stream().map(AnalyticsEvent::id).toList());
    }

    // Opting out empties the queue rather than only stopping new writes.
    // Events already on disk were collected under a consent the user has just
    // withdrawn, and shipping them anyway is the thing the opt out was for.
    public synchronized void clear() {
        while (storage.size() > 0) {
            List<AnalyticsEvent> batch = storage.oldest(batchSize);
            if (batch.isEmpty()) {
                return;
            }
            storage.delete(batch.stream().map(AnalyticsEvent::id).toList());
        }
    }

    public synchronized int droppedCount() {
        return dropped;
    }
}

package com.androidinterview.analytics;

import java.util.Map;
import java.util.UUID;

// Who the event belongs to. Three ids, and each one exists because a report
// breaks without it. The anonymous id is the install, the user id is the
// person, and the session id is one visit.
public final class Identity {

    // The anonymous id is persisted on first launch and read back on every
    // launch after that. An id regenerated per process turns one user into a
    // new user every cold start, and every retention number is then wrong.
    public interface Store {
        String anonymousId();

        void setAnonymousId(String anonymousId);

        String userId();

        void setUserId(String userId);
    }

    private final Store store;
    private final long sessionTimeoutMillis;

    private String sessionId = UUID.randomUUID().toString();
    private long lastEventAtMillis;

    public Identity(Store store, long sessionTimeoutMillis) {
        this.store = store;
        this.sessionTimeoutMillis = sessionTimeoutMillis;
        if (store.anonymousId() == null) {
            store.setAnonymousId(UUID.randomUUID().toString());
        }
    }

    public String anonymousId() {
        return store.anonymousId();
    }

    public String userId() {
        return store.userId();
    }

    // A session is a gap, not a lifecycle callback. Anything more than the
    // timeout since the last event starts a new one, which is why this is
    // driven by event time rather than by onStart. A user who leaves the app
    // open in the background for an hour comes back to a new session.
    public synchronized String sessionId(long nowMillis) {
        if (nowMillis - lastEventAtMillis > sessionTimeoutMillis) {
            sessionId = UUID.randomUUID().toString();
        }
        lastEventAtMillis = nowMillis;
        return sessionId;
    }

    // Login. The alias event is the one that saves the pre-login funnel. It
    // tells the backend that everything ever sent under this anonymous id is
    // the same person as this user id, so a signup can still be attributed to
    // the ad that brought the install in. The anonymous id is kept, because
    // rotating it here would orphan exactly the events we are trying to join.
    public synchronized AnalyticsEvent login(String userId, long nowMillis) {
        String previous = store.anonymousId();
        store.setUserId(userId);
        return new AnalyticsEvent(
                UUID.randomUUID().toString(),
                "$alias",
                Map.of("previous_id", previous, "user_id", userId),
                nowMillis,
                previous,
                userId,
                sessionId(nowMillis));
    }

    // Logout. A fresh anonymous id and a fresh session, because the next
    // person holding this phone is not the last one. Keeping the id would
    // stitch two people into one profile, which is worse than losing the link.
    public synchronized void logout() {
        store.setUserId(null);
        store.setAnonymousId(UUID.randomUUID().toString());
        sessionId = UUID.randomUUID().toString();
        lastEventAtMillis = 0;
    }
}

package com.androidinterview.analytics

// Asking WorkManager for a one time flush. Keeping it a seam is what lets this
// class be tested without Android, and it is also where the unique work name
// goes, so two requests never run two workers over one queue.
fun interface FlushScheduler {
    fun requestFlush()
}

// The only thing a call site sees. track() validates, stamps identity, writes
// to disk and returns. No network call, no lock held on anything slow, and
// nothing that can throw in a release build.
class Analytics(
    // The schema, event name to the properties that event must carry. This is
    // the thing a logging library has no reason to own. A typo in an event name
    // is invisible until someone builds a dashboard on it a month later, so the
    // library refuses the event at the call site instead.
    private val schema: Map<String, Set<String>>,
    private val destinations: List<EventQueue>,
    private val identity: Identity,
    private val scheduler: FlushScheduler,
    private val now: () -> Long = System::currentTimeMillis,
    // Debug builds throw so a bad event fails in review. Release builds drop
    // and count, because an analytics library must never crash the feature it
    // is measuring.
    private val strict: Boolean = false,
) {

    @Volatile
    private var enabled = true

    var rejectedCount = 0
        private set

    // The consent gate sits in front of the queue, not in front of the
    // uploader. Opting out drops what is already on disk, because those events
    // were collected under a consent that no longer exists.
    fun setTrackingEnabled(value: Boolean) {
        enabled = value
        if (!value) destinations.forEach(EventQueue::clear)
    }

    fun track(name: String, properties: Map<String, Any> = emptyMap()) {
        if (!enabled) return
        if (schema[name]?.all { it in properties.keys } != true) {
            rejectedCount++
            require(!strict) { "Event does not match the schema, $name" }
            return
        }
        deliver(eventOf(name, properties, identity, now()))
    }

    fun login(userId: String) {
        if (enabled) deliver(identity.login(userId, now()))
    }

    fun logout() = identity.logout()

    // One of the three flush triggers. The app going to the background is the
    // most valuable one, because it is the last moment before the process can
    // be killed, and the events from the session that just ended are sitting in
    // the queue.
    fun onAppBackgrounded() = scheduler.requestFlush()

    // Fan out. The call site names the event once and every configured
    // destination gets its own copy in its own queue, so a third party backend
    // being down cannot hold up the first party one.
    private fun deliver(event: AnalyticsEvent) {
        // map, not any, because any short circuits and the second destination
        // would never see the event.
        if (destinations.map { it.append(event) }.any { it }) scheduler.requestFlush()
    }
}

package com.androidinterview.analytics

import java.util.UUID

// One tracked event. The id is minted on the device inside track() and never
// changes, not even across a retry, and that is the whole idempotency story. A
// batch the server accepted whose acknowledgement was lost gets sent again,
// and every event in it arrives carrying the id it already had, so the backend
// drops the duplicate instead of double counting the funnel.
data class AnalyticsEvent(
    val name: String,
    val properties: Map<String, Any>,
    // The device clock at track() time. The server stamps its own receive time
    // beside it, because a phone with a wrong clock is common and the backend
    // needs both numbers to notice.
    val timestampMillis: Long,
    val anonymousId: String,
    // Null until the user logs in. Everything before that is stitched by the
    // alias event Identity emits.
    val userId: String?,
    val sessionId: String,
    val id: String = UUID.randomUUID().toString(),
)

internal fun eventOf(
    name: String,
    properties: Map<String, Any>,
    identity: Identity,
    nowMillis: Long,
) = AnalyticsEvent(
    name = name,
    properties = properties,
    timestampMillis = nowMillis,
    anonymousId = identity.anonymousId,
    userId = identity.userId,
    sessionId = identity.sessionId(nowMillis),
)

package com.androidinterview.analytics

import java.io.IOException
import java.util.UUID

// Three outcomes, not two. Accepted and a transient failure are obvious, and
// the third one is the one people forget. A batch the server rejects outright,
// a schema the backend no longer accepts, retries forever and blocks every
// event behind it, so it is dropped on purpose.
enum class Outcome { ACCEPTED, TRANSIENT_FAILURE, REJECTED }

fun interface Transport {
    // Declared as throwing, because a dropped connection is the ordinary case
    // here and the caller treats it exactly like a transient failure.
    @Throws(IOException::class)
    fun send(batchId: String, events: List<AnalyticsEvent>): Outcome
}

// The batch id is derived from the ids of the events in the batch rather than
// minted per attempt. A retry of the same events therefore carries the same
// batch id, so the server can recognise the replay in one lookup instead of
// deduplicating event by event.
internal fun batchIdOf(events: List<AnalyticsEvent>): String =
    UUID.nameUUIDFromBytes(events.joinToString(",") { it.id }.toByteArray()).toString()

// What the background worker calls. Everything here runs off the calling
// thread, so a track() on the main thread never waits on a socket.
class BatchUploader(
    private val queue: EventQueue,
    private val transport: Transport,
) {

    // Returns true when the queue was drained. False means a transient
    // failure, and the caller turns that into WorkManager's retry, which owns
    // the backoff. Writing a second backoff timer in here would just fight it.
    fun flush(): Boolean {
        while (true) {
            val batch = queue.nextBatch()
            if (batch.isEmpty()) return true
            val outcome = try {
                transport.send(batchIdOf(batch), batch)
            } catch (networkFailure: IOException) {
                return false
            }
            if (outcome == Outcome.TRANSIENT_FAILURE) return false
            // A rejected batch is acknowledged too, which reads wrong for a
            // second and is right. Leaving it queued means a poison batch that
            // nothing behind it can get past.
            queue.acknowledge(batch)
        }
    }
}

package com.androidinterview.analytics

// Room, or an append only file. The queue does not care which, it only needs
// the rows back in the order they went in.
interface EventStorage {
    fun append(event: AnalyticsEvent)
    fun oldest(limit: Int): List<AnalyticsEvent>
    fun delete(eventIds: List<String>)
    val size: Int
}

// The queue for one destination. It is persisted, because an in memory list
// loses every event tracked since the last flush the moment the process dies,
// and the screens users bounce off fastest are the ones that die first.
//
// One queue per destination, not one queue with a destination column. A
// backend that is down then backs up only its own queue, and the healthy
// destination keeps draining.
class EventQueue(
    val destination: String,
    private val storage: EventStorage,
    private val batchSize: Int = 20,
    private val capacity: Int = 10_000,
) {

    var droppedCount = 0
        private set

    // Returns true when the count trigger has fired, which is one of the three
    // reasons a flush happens. The other two are the periodic worker and the
    // app going to the background, and both live outside this class because
    // they are lifecycle, not queueing.
    //
    // The cap is the important line here. A user offline for a week must not
    // come back to fifty thousand queued events and one flush that hammers the
    // radio, so the oldest event is dropped to make room and the drop is
    // counted. Reporting the count matters more than it looks, because a
    // silent drop shows up later as a funnel that quietly lost its top.
    @Synchronized
    fun append(event: AnalyticsEvent): Boolean {
        if (storage.size >= capacity) {
            storage.oldest(1).firstOrNull()?.let {
                storage.delete(listOf(it.id))
                droppedCount++
            }
        }
        storage.append(event)
        return storage.size >= batchSize
    }

    @Synchronized
    fun nextBatch(): List<AnalyticsEvent> = storage.oldest(batchSize)

    // Deleting happens only after the server has acknowledged the batch. The
    // cost of that order is a duplicate when the acknowledgement is lost, and
    // the event id is what makes that duplicate free to throw away.
    @Synchronized
    fun acknowledge(batch: List<AnalyticsEvent>) = storage.delete(batch.map { it.id })

    // Opting out empties the queue rather than only stopping new writes.
    // Events already on disk were collected under a consent the user has just
    // withdrawn, and shipping them anyway is the thing the opt out was for.
    @Synchronized
    fun clear() {
        while (true) {
            val batch = storage.oldest(batchSize)
            if (batch.isEmpty()) return
            storage.delete(batch.map { it.id })
        }
    }
}

package com.androidinterview.analytics

import java.util.UUID

// The anonymous id is persisted on first launch and read back on every launch
// after that. An id regenerated per process turns one user into a new user
// every cold start, and every retention number is then wrong.
interface IdentityStore {
    var anonymousId: String?
    var userId: String?
}

// Who the event belongs to. Three ids, and each one exists because a report
// breaks without it. The anonymous id is the install, the user id is the
// person, and the session id is one visit.
class Identity(
    private val store: IdentityStore,
    private val sessionTimeoutMillis: Long = 30 * 60 * 1000,
) {

    private var currentSessionId = UUID.randomUUID().toString()
    private var lastEventAtMillis = 0L

    init {
        if (store.anonymousId == null) store.anonymousId = UUID.randomUUID().toString()
    }

    val anonymousId: String get() = requireNotNull(store.anonymousId)

    val userId: String? get() = store.userId

    // A session is a gap, not a lifecycle callback. Anything more than the
    // timeout since the last event starts a new one, which is why this is
    // driven by event time rather than by onStart. A user who leaves the app
    // open in the background for an hour comes back to a new session.
    @Synchronized
    fun sessionId(nowMillis: Long): String {
        if (nowMillis - lastEventAtMillis > sessionTimeoutMillis) {
            currentSessionId = UUID.randomUUID().toString()
        }
        lastEventAtMillis = nowMillis
        return currentSessionId
    }

    // Login. The alias event is the one that saves the pre-login funnel. It
    // tells the backend that everything ever sent under this anonymous id is
    // the same person as this user id, so a signup can still be attributed to
    // the ad that brought the install in. The anonymous id is kept, because
    // rotating it here would orphan exactly the events we are trying to join.
    @Synchronized
    fun login(userId: String, nowMillis: Long): AnalyticsEvent {
        val previous = anonymousId
        store.userId = userId
        return AnalyticsEvent(
            name = "\$alias",
            properties = mapOf("previous_id" to previous, "user_id" to userId),
            timestampMillis = nowMillis,
            anonymousId = previous,
            userId = userId,
            sessionId = sessionId(nowMillis),
        )
    }

    // Logout. A fresh anonymous id and a fresh session, because the next person
    // holding this phone is not the last one. Keeping the id would stitch two
    // people into one profile, which is worse than losing the link.
    @Synchronized
    fun logout() {
        store.userId = null
        store.anonymousId = UUID.randomUUID().toString()
        currentSessionId = UUID.randomUUID().toString()
        lastEventAtMillis = 0L
    }
}