//! Backend logic to upload a document to multiple targets

use std::{

    collections::HashMap,

    fmt::Debug,

    hash::Hash,

    num::NonZeroUsize,

    sync::{

        Arc,

        atomic::{self, AtomicUsize},

    },

    time::Duration,

};

use futures::{

    FutureExt as _, StreamExt as _, future::BoxFuture, select_biased, stream::Fuse,

    stream::FuturesUnordered,

};

use postage::watch;

use tor_basic_utils::retry::RetryDelay;

use tor_error::warn_report;

use tor_rtcompat::SleepProvider;

use tracing::{Level, debug, span, trace, warn};

use web_time_compat::Instant;

use crate::{

    DocVersion, Document, PublishDirective, PublishStatus, Rejection, UploadError, Uploader,

};

/// Identifier for a single action that we have queued for a target.

///

/// (Actions can currently be "try to upload" or "wait till later.")

#[derive(Eq, PartialEq, Clone, Copy, Debug)]

struct ActionNum(NonZeroUsize);

impl ActionNum {

    /// Return a new identifier.

    ///

    /// We do not guarantee that these are permanently unique:

    /// only that there are very unlikely to be two actions with the same ActionNum

    /// active at once.

        static NEXT: AtomicUsize = AtomicUsize::new(0);

        loop {

        }

}

/// Mutable status for a single target.

#[derive(Debug)]

struct TargetStatus {

    /// The current state of this target.

    state: TargetState,

    /// How many times has this target failed since we last received an answer from it?

    /// ("Published" and "Rejected" both count as answers; "Try again later" does not.)

    n_failures: usize,

    /// State of our retry-timing algorithm.

    retry: RetryDelay,

    /// Identifier for the most recent action that we launched for this target.

    ///

    /// (Multiple actions can be pending at once.

    /// We call an action "the latest action for a target"

    /// if its ActionNum matches this value.)

    latest_action: Option<ActionNum>,

}

/// The current state for a single target.

#[derive(Debug)]

#[expect(unused)] // TODO: we don't use all these fields yet; we should remove or expose them.

enum TargetState {

    /// There is no document to upload, so we don't have anything to do.

    NoDocument,

    /// We are ready to try uploading the most recent document to this target.

    Ready,

    /// We are trying to upload the most recent document to this target.

    ///

    /// (Invariant: If a target is in this state, there is an Upload future

    /// among our pending actions for that target.)

    Inflight {

        /// The time when we began trying to upload.

        since: Instant,

    },

    /// An upload attempt has failed; we are waiting for a while before we try again.

    ///

    /// (Invariant: If a target is in this state, there is a Sleep future

    /// among our pending actions for that target.)

    Waiting {

        /// The time until which we are waiting.

        until: Instant,

    },

    /// We have published the most recent document to this target.

    Published,

    /// We have failed permanently for some reason.

    /// We won't retry until the document changes.

    PermanentlyFailed(UploadError),

    /// The target told us that it will not accept the most recent document,

    /// and so we should not try that document again.

    Rejected(Rejection),

}

/// An enum to declare whether a document is present.

//

// (We don't use bool here because bools are error-prone.)

#[derive(Clone, Copy, Debug)]

enum DocumentPresent {

    /// We have a document.

    Present,

    /// We do not currently have a document to upload

    Absent,

}

impl DocumentPresent {

    /// Return the initial state that new targets should enter.

        }

}

impl<D: ?Sized> super::Document<D> {

    /// Return a DocumentPresent for this document.

        } else {

        }

}

impl TargetStatus {

    /// Construct a new TargetStatus in the Ready state.

    /// Called after the latest action for a target has failed:

    /// increments the failure count, and returns the interval for which we should wait.

    ///

    /// If `suggested_delay` is provided, it is an amount that the target

    /// told us to wait before retrying.   We will treat this amount

    /// as a _minimum_.  A `suggested_delay` will not prevent us

    /// from retrying immediately if our failure status is reset.

    /// Called after we have decided to launch an upload for a target.

    /// Change this action's state to Ready.

    /// Reset the failure count and timeout for this target.

    /// Called after the latest action for a target has succeeded

    /// in uploading the most recent document.

    /// Called after the latest action for a target has been rejected

    /// in uploading the most recent document.

    /// Mark this target as permanently unable to receive the current

    /// document because of some error `e`.

}

/// The result of a single action.

#[derive(Debug)]

enum ActionOutcome {

    /// A sleep action has expired.

    DoneSleeping,

    /// An upload action has succeeded.

    Published,

    /// An upload action has been rejected.

    Rejected(Rejection),

    /// An upload action has failed with some error.

    Err(UploadError),

}

impl ActionOutcome {

    /// Construct an [`ActionOutcome`] from the result of [`Uploader::upload()`].

        }

}

/// The type returned by one of the action futures in `PublishReactor.inflight`.

struct TaskResult<T: ?Sized> {

    /// Which target were we taking this action for?

    target: Arc<T>,

    /// An `ActionNum` to identify whether the action was the latest one for the target.

    action: ActionNum,

    /// Which document was the most recent when the action was launched?

    doc_version: DocVersion,

    /// The result of the action.

    outcome: ActionOutcome,

}

/// Backend data that we use to publish a document (or series of documents).

pub(crate) struct PublishReactor<R: SleepProvider, D: ?Sized, T, UP: ?Sized>

where

    T: Hash + Eq + ?Sized,

{

    /// A sleep provider used to launch wait actions.

    runtime: R,

    /// A description of what we're uploading, for log messages.

    description: String,

    /// The current actions that the [`Publisher`](crate::Publisher) has told us to take.

    ///

    /// We watch for changes in this directive and adjust our behavior accordingly.

    directive: Fuse<watch::Receiver<PublishDirective<D, T>>>,

    /// A channel we use to report our current status to the [`Publisher`](crate::Publisher).

    status: watch::Sender<PublishStatus>,

    /// The document which we are currently trying to upload.

    latest_document: Document<D>,

    /// The initial retry delay for a failed target.

    ///

    /// (Used to seed our retry delay algorithm.)

    initial_retry_delay: Duration,

    /// The most recent value we've seen for the `reset_count` field of our `PublishDirective`.

    latest_reset_count: usize,

    /// The [`Uploader`] object we use to upload documents.

    uploader: Arc<UP>,

    /// A set of all our pending actions.

    ///

    /// Actions are either upload attempts, or sleep actions.

    ///

    /// Additionally, this `FuturesUnordered` contains a single future that is always

    /// pending, to guarantee that `inflight.next()` never returns None.

    ///

    /// Multiple actions may be inflight for a given target at a time.

    /// This is deliberate:

    /// If we change our document while the upload of an older document is inflight,

    /// we do not want to stop the inflight upload in the middle.

    ///

    /// TODO: We _could_ cancel any Sleep action that is superseded.

    /// That's a fair amount of effort, though, since FuturesUnordered doesn't have

    /// very nice accessors nor does Sleep have a good way to make it cancellable.

    inflight: FuturesUnordered<BoxFuture<'static, TaskResult<T>>>,

    /// The current status for each of our live targets.

    target_status: HashMap<Arc<T>, TargetStatus>,

}

/// Return type used to tell the reactor loop to exit.

#[derive(Debug)]

struct ExitLoop;

impl<R: SleepProvider, D, T, UP> PublishReactor<R, D, T, UP>

where

    D: ?Sized,

    T: Hash + Eq + Send + Sync + Debug + ?Sized + 'static,

    UP: Uploader<Doc = D, Target = T> + ?Sized,

{

    /// Construct a new reactor.

    ///

    /// (Does not launch any background task or do any work).

        // Add a never-finished future to keep FuturesUnordered from saying it's done.

    /// Run forever, handling changes in the [`PublishDirective`], uploading documents, and reporting status.

        // The first time we start, we begin uploading.

        'mainloop: loop {

                // We've been told to do something new, _or_ the last handle to the Publisher has

                // been dropped.

                        // The watch::Receiver stream returned None,

                        // so we know that the last handle has been dropped.

                    };

                    // Process any change in the action.

                    // Update our `PublishStatus`.

                }

                // Some action has finished; update accordingly.

            }

        }

    /// Called when a task in `self.inflight` produces a result.

    ///

    /// Update our status and launch new tasks as appropriate.

    #[allow(clippy::cognitive_complexity)]

        let TaskResult {

            // The target isn't here, so we don't care about what happened with it.

        };

            // There is a more recent inflight action for this target;

            // ignore the results of this one.

            //

            // (See note on `Publish.inflight` about why we can have multiple inflight

            // actions.)

            //

            // We use a != comparison here rather than < since we allow the action

            // identifier space to wrap around.

            ActionOutcome::Published => {

                    // We aren't tracking this particular document any more;

                    // this was a stale upload.

            }

                    // We aren't tracking this particular document any more;

                    // this was a stale upload.

                    "{} upload rejected. The target ({:?}) said {}",

                );

            }

                    "Attempt to publish {} to {:?} failed. Not retriable.",

                );

            }

                // We failed to upload: we log the error and wait until it's time to

                // retry.

                // TODO: This might need to be downgraded, but for now we'll leave it as-is.

                    "Attempt to publish {} to {:?} failed. We'll retry later.",

                );

            }

        }

    /// Called when we have received a new [`PublishDirective`] from the publisher.

    ///

    /// Update the status of all of our targets, and launch new uploads as appropriate.

        use TargetState::*;

            // We're supposed to shut down.  Just go ahead and do that.

        // Check to see if any targets have been added or removed;

        // update target_status accordingly.

        }

        // Have gotten a new document?  Have we been told to reset failing targets?

        //

        // (We use != rather than > here since we want to allow these counters to wrap around.)

        // Update our own versions of the counters from the PublishDirective.

            } else {

            }

        // Reset failure timings if appropriate,

        // and mark targets ready if we want to launch a new upload to them.

                // If this target is waiting, then we should let it continue

                // waiting unless we've been told to reset failing targets.

                // If the target is ready, there's no point in making it ready.

                // If we're currently uploading to a target,

                // we only want to launch a new upload if the document changed.

                // If we've published successfully,

                // or if we have been rejected,

                // or if we had nothing to do,

                // we only want to launch a new upload if the document changed.

                // If we had no document, we want to launch now that we have one.

            };

        }

    /// Launch a new upload for every Ready target,

    /// making its status Inflight.

        // Build a list of the ready targets.

        //

        // This is a separate step to avoid a concurrent mutable/immutable borrow.

        // Launch an upload for each of them.

    /// Compute a new [`PublishStatus`] reflecting our progress uploading the current document,

    /// and deliver it to the Publisher.

        use TargetState::*;

                Inflight { .. } | Waiting { .. } => {

                }

            }

        }

    /// Launch an upload action for a given `target`, changing its status to Inflight.

        // Launch the publish request, and add it to inflight.

        };

            // There's no document, so we can't upload it.

        };

    /// Launch a sleep action for a given `target`, changing its status to `Waiting`.

        // Launch the publish request, and add it to inflight.

        };

            ?target,

            ?delay,

            "Waiting for next {} upload attempt.",

        );

}