ores.marketdata.market_observation

std::vector<domain::market_observation> read_latest_for_series(context ctx,
                                                               const boost::uuids::uuid& series_id);

#include "ores.database/repository/unit_of_work.hpp"
#include "ores.marketdata.api/datum/series_stream.hpp"
#include "ores.marketdata.core/repository/as_of_rows.hpp"
#include "ores.marketdata.core/repository/market_series_repository.hpp"
#include "ores.marketdata.core/repository/observation_lineage_repository.hpp"
#include "ores.marketdata.core/repository/pg_array.hpp"
#include "ores.marketdata.core/repository/tick_entitlement.hpp"
#include "ores.platform/time/datetime.hpp"
#include <boost/uuid/nil_generator.hpp>
#include <boost/uuid/random_generator.hpp>
#include <boost/uuid/uuid_io.hpp>
#include <format>
#include <map>
#include <optional>
#include <set>
#include <string>
#include <utility>
#include <vector>

std::vector<domain::market_observation>
market_observation_repository::read_latest_for_series(context ctx,
                                                     const boost::uuids::uuid& series_id) {
    BOOST_LOG_SEV(lg(), debug) << "Reading latest observations for series: " << series_id;
    static constexpr auto max = db_timestamp::max();
    const auto tid = ctx.tenant_id().to_string();
    const auto sid = boost::uuids::to_string(series_id);
    const auto query =
        query_builder::read<std::vector<market_observation_entity>> |
        query_builder::where("tenant_id"_c == tid && "series_id"_c == sid && "valid_to"_c == max) |
        query_builder::order_by("observation_datetime"_c);

    return query_builder::execute_read_query<market_observation_entity, domain::market_observation>(
        ctx,
        query,
        [](const auto& entities) { return market_observation_mapper::map(entities); },
        lg(),
        "Reading latest market observations.");
}

/**
 * @brief Every current observation of the tenant, ordered by series and time.
 *
 * One statement for the whole tenant. A caller that needs the observations of many
 * series asks here, because a statement for each series pays the planner once for
 * every chunk of the hypertable. The rows are held in memory together, so the size
 * of the answer grows with the history the tenant keeps.
 */
std::vector<domain::market_observation> read_latest_all(context ctx);

std::vector<domain::market_observation>
market_observation_repository::read_latest_all(context ctx) {
    BOOST_LOG_SEV(lg(), debug) << "Reading the latest observations of every series";
    static constexpr auto max = db_timestamp::max();
    const auto tid = ctx.tenant_id().to_string();
    const auto query = query_builder::read<std::vector<market_observation_entity>> |
                       query_builder::where("tenant_id"_c == tid && "valid_to"_c == max) |
                       query_builder::order_by("series_id"_c, "observation_datetime"_c);

    return query_builder::execute_read_query<market_observation_entity, domain::market_observation>(
        ctx,
        query,
        [](const auto& entities) { return market_observation_mapper::map(entities); },
        lg(),
        "Reading the latest market observations of every series.");
}

// Declared here and defined in as_of_rows.hpp. The header that declares this
// repository needs the name only, and the definition sits beside the read that
// builds it, so a caller that wants the record times includes that one.
struct observation_record;

std::vector<observation_record>
read_as_of_records(context ctx,
                   const boost::uuids::uuid& series_id,
                   const std::chrono::system_clock::time_point& as_of_datetime);

namespace {

/*
 * The stream of the ticks of one series, as the ingest derives it from the provider and the
 * series URI. The nil id returns when the session cannot see the series, or when it names a
 * party that may not read the provider's ticks. No tick carries the nil id, so the read finds no
 * tick and still answers with the party's own observation rows. A context with no party is a
 * service context, which reads the store on its own authority, as every service read does. The
 * id is stored nowhere, so every read computes it. @p allowed is the right the read needs.
 */
std::string stream_of(market_observation_repository::context ctx,
                                     const boost::uuids::uuid& series_id,
                                     bool (*allowed)(const ores::database::context&, const std::string&)) {
    const auto series =
        market_series_repository{}.read_latest(ctx, boost::uuids::to_string(series_id));
    if (series.empty())
        return boost::uuids::to_string(boost::uuids::nil_uuid());
    const auto& s = series.front();
    if (ctx.party_id() && !allowed(ctx, s.provider_code))
        return boost::uuids::to_string(boost::uuids::nil_uuid());
    return boost::uuids::to_string(datum::stream_id(s.provider_code, s.oresmd_uri));
}

/*
 * The stream of every series of the tenant that the session may compute from, with the
 * series each belongs to, in one order. read_as_of_all reads every series in one statement, and
 * no statement derives the id.
 */
std::pair<std::vector<std::string>, std::vector<std::string>>
streams_of_every_series(market_observation_repository::context ctx) {
    const auto party = ctx.party_id();
    const auto usable = party ? computable_providers(ctx) : std::set<std::string>{};
    std::vector<std::string> series;
    std::vector<std::string> streams;
    for (const auto& s : market_series_repository{}.read_latest(ctx)) {
        if (party && !usable.contains(s.provider_code))
            continue;
        series.push_back(boost::uuids::to_string(s.id));
        streams.push_back(boost::uuids::to_string(datum::stream_id(s.provider_code, s.oresmd_uri)));
    }
    return {std::move(series), std::move(streams)};
}

/*
 * The datums of one series' stream, with the owner the tick rows no longer carry. The stream id
 * is derived rather than stored, so the caller states it in @p stream. $2 is the series, and
 * @p max is the record time the current series row carries.
 */
std::string tick_coords_cte(std::string_view stream, std::string_view max) {
    return std::format(R"(coords AS (
            SELECT c.stream_id, c.coord, c.oresmd_uri, s.id AS series_id, s.tenant_id, s.party_id
            FROM ores_marketdata_tick_coords_tbl c
            JOIN ores_marketdata_market_series_tbl s
              ON s.id = $2::uuid AND s.valid_to = {1}::timestamptz
            WHERE c.stream_id = {0}::uuid
        ))",
                       stream,
                       max);
}

/*
 * The latest tick of each coordinate at or before @p bound, as the ten observation columns and a
 * source rank of 1. One index probe finds each, so a long series costs a few rows. A tick has
 * no source, and its record time is its own time. Needs the coords CTE.
 */
std::string latest_tick_points(std::string_view bound, std::string_view max) {
    return std::format(R"(
            SELECT md5(c.series_id::text || t.coord::text || t.ts::text)::uuid,
                   c.tenant_id, c.party_id, c.series_id, t.ts, c.oresmd_uri, t.value::text,
                   NULL::text, t.feed_id, t.ts, {1}::timestamptz, 1
            FROM coords c
            CROSS JOIN LATERAL (
                SELECT * FROM ores_marketdata_ticks_tbl x
                WHERE x.stream_id = c.stream_id AND x.coord = c.coord
                    AND x.ts <= {0}::timestamptz
                ORDER BY x.ts DESC
                LIMIT 1
            ) t)",
                       bound,
                       max);
}

/*
 * The manual points that own their coordinate for reading: the observation
 * each current manual annex row of this series keyed, with that row's own
 * record time. The annex is cold and holds only hand-keyed points, so this
 * read is small; keeping it separate from the observation scan leaves the
 * tick path unchanged.
 */
std::vector<observation_record>
read_manual_points(market_observation_repository::context ctx,
                   const boost::uuids::uuid& series_id,
                   ores::logging::logger_t& log) {
    const auto tid = ctx.tenant_id().to_string();
    const auto sid = boost::uuids::to_string(series_id);
    static const std::string sql = R"(
        SELECT o.id, o.tenant_id, o.party_id, o.series_id, o.observation_datetime,
               o.oresmd_uri, o.value, o.source, o.feed_id, o.valid_from, o.valid_to
        FROM ores_marketdata_observation_lineages_tbl l
        JOIN ores_marketdata_market_observations_tbl o
          ON o.tenant_id = l.tenant_id AND o.party_id = l.party_id
         AND o.series_id = l.series_id AND o.oresmd_uri = l.oresmd_uri
         AND o.observation_datetime = l.observation_datetime AND o.valid_to = $3
        WHERE l.tenant_id = $1 AND l.series_id = $2 AND l.point_source_kind = 'manual'
            AND l.valid_to = $3
    )";

    const auto rows = query_builder::execute_parameterized_multi_column_query(
        ctx, sql, {tid, sid, MAX_TIMESTAMP}, log, "reading manual curve points");

    std::vector<observation_record> result;
    result.reserve(rows.size());
    for (const auto& row : rows)
        result.push_back({market_observation_mapper::map(
                              as_of_observation(row, 0, "read_manual_points")),
                          as_of_recorded_at(row, 0, "read_manual_points")});
    return result;
}

/*
 * The manual points of every series of the tenant: the read above without its series.
 */
std::vector<observation_record> read_manual_points_all(market_observation_repository::context ctx,
                                                       ores::logging::logger_t& log) {
    const auto tid = ctx.tenant_id().to_string();
    static const std::string sql = R"(
        SELECT o.id, o.tenant_id, o.party_id, o.series_id, o.observation_datetime,
               o.oresmd_uri, o.value, o.source, o.feed_id, o.valid_from, o.valid_to
        FROM ores_marketdata_observation_lineages_tbl l
        JOIN ores_marketdata_market_observations_tbl o
          ON o.tenant_id = l.tenant_id AND o.party_id = l.party_id
         AND o.series_id = l.series_id AND o.oresmd_uri = l.oresmd_uri
         AND o.observation_datetime = l.observation_datetime AND o.valid_to = $2
        WHERE l.tenant_id = $1 AND l.point_source_kind = 'manual'
            AND l.valid_to = $2
    )";

    const auto rows = query_builder::execute_parameterized_multi_column_query(
        ctx, sql, {tid, MAX_TIMESTAMP}, log, "reading manual curve points of every series");

    std::vector<observation_record> result;
    result.reserve(rows.size());
    for (const auto& row : rows)
        result.push_back(
            {market_observation_mapper::map(as_of_observation(row, 0, "read_manual_points_all")),
             as_of_recorded_at(row, 0, "read_manual_points_all")});
    return result;
}

}

std::vector<observation_record> market_observation_repository::read_as_of_records(
    context ctx,
    const boost::uuids::uuid& series_id,
    const std::chrono::system_clock::time_point& as_of_datetime) {
    using ores::platform::time::datetime;
    BOOST_LOG_SEV(lg(), debug) << "Reading as-of snapshot for series: " << series_id
                               << " as-of: " << datetime::to_iso8601_utc(as_of_datetime);
    const auto tid = ctx.tenant_id().to_string();
    const auto sid = boost::uuids::to_string(series_id);
    const auto as_of_str = datetime::to_iso8601_utc(as_of_datetime);
    const auto stream = stream_of(ctx, series_id, may_compute_from_ticks);

    // DISTINCT ON (oresmd_uri) returns exactly one row per point -- the latest at or before
    // as_of_datetime -- reconstructing a curve/grid snapshot from independently-ticking rows.
    // Correct whether every point shares one observation_datetime or the points have
    // staggered timestamps.
    //
    // The points come from two tables: the observation table, and the tick table through the
    // coordinate dictionary. The tick branch takes the latest tick of each coordinate with one
    // index probe, so it reads a few rows however long the series is. On a tie the observation
    // row wins, because it carries the source and its own id.
    static const std::string sql = "WITH " + tick_coords_cte("$5", "$4") + R"(,
        points AS (
            SELECT id, tenant_id, party_id, series_id, observation_datetime, oresmd_uri, value,
                   source, feed_id, valid_from, valid_to, 0 AS src_rank
            FROM ores_marketdata_market_observations_tbl
            WHERE tenant_id = $1 AND series_id = $2 AND observation_datetime <= $3
                AND valid_to = $4
            UNION ALL)" + latest_tick_points("$3", "$4") +
                                   R"(
        )
        SELECT DISTINCT ON (oresmd_uri)
            id, tenant_id, party_id, series_id, observation_datetime, oresmd_uri, value,
            source, feed_id, valid_from, valid_to
        FROM points
        ORDER BY oresmd_uri, observation_datetime DESC, src_rank
    )";

    const auto rows = query_builder::execute_parameterized_multi_column_query(
        ctx,
        sql,
        {tid, sid, as_of_str, MAX_TIMESTAMP, stream},
        lg(),
        "reading as-of curve snapshot");

    std::vector<observation_record> result;
    result.reserve(rows.size());
    for (const auto& row : rows)
        result.push_back({market_observation_mapper::map(
                              as_of_observation(row, 0, "read_as_of_records")),
                          as_of_recorded_at(row, 0, "read_as_of_records")});

    // A current manual point owns its coordinate until cleared, overlaying the
    // fed value and any later automatic write. One whose instant is after the
    // snapshot instant is not in it; one the snapshot covers is, whenever keyed.
    for (const auto& manual : read_manual_points(ctx, series_id, lg())) {
        if (manual.observation.observation_datetime > as_of_datetime)
            continue;
        const auto it = std::ranges::find(
            result, manual.observation.oresmd_uri, [](const observation_record& record) {
                return record.observation.oresmd_uri;
            });
        if (it != result.end())
            *it = manual;
        else
            result.push_back(manual);
    }
    return result;
}

std::vector<domain::market_observation>
read_as_of(context ctx,
           const boost::uuids::uuid& series_id,
           const std::chrono::system_clock::time_point& as_of_datetime);

std::vector<domain::market_observation> market_observation_repository::read_as_of(
    context ctx,
    const boost::uuids::uuid& series_id,
    const std::chrono::system_clock::time_point& as_of_datetime) {
    const auto records = read_as_of_records(ctx, series_id, as_of_datetime);
    std::vector<domain::market_observation> result;
    result.reserve(records.size());
    for (const auto& record : records)
        result.push_back(record.observation);
    return result;
}

/**
 * @brief The as-of snapshot of every series of the tenant, in one statement.
 *
 * Each series gives what @c read_as_of gives for it: the latest point of each datum at
 * or before the instant, from the observation and tick tables, with a current manual
 * point owning its datum. A row carries its series in @c series_id. The statement
 * is that of @c read_as_of_records without its series filter, and a change to one
 * is a change to the other.
 */
std::vector<domain::market_observation>
read_as_of_all(context ctx, const std::chrono::system_clock::time_point& as_of_datetime);

std::vector<domain::market_observation> market_observation_repository::read_as_of_all(
    context ctx, const std::chrono::system_clock::time_point& as_of_datetime) {
    using ores::platform::time::datetime;
    const auto tid = ctx.tenant_id().to_string();
    const auto as_of_str = datetime::to_iso8601_utc(as_of_datetime);
    BOOST_LOG_SEV(lg(), debug) << "Reading the as-of snapshot of every series as-of: " << as_of_str;

    // The streams of the tenant travel as two arrays of one order, because the tick rows name
    // their stream by a derived id and no statement derives it. The statement of
    // read_as_of_records without its series filter: the coordinates of every series, the latest
    // tick of each by one index probe, and the observation rows, reduced to one row for each
    // datum of each series.
    const auto [series_ids, streams] = streams_of_every_series(ctx);

    static const std::string sql = R"(
        WITH coords AS (
            SELECT x.series_id, x.stream_id, s.tenant_id, s.party_id, c.coord, c.oresmd_uri
            FROM unnest($4::uuid[], $5::uuid[]) AS x(series_id, stream_id)
            JOIN ores_marketdata_market_series_tbl s
              ON s.id = x.series_id AND s.valid_to = $3
            JOIN ores_marketdata_tick_coords_tbl c ON c.stream_id = x.stream_id
        ),
        points AS (
            SELECT id, tenant_id, party_id, series_id, observation_datetime, oresmd_uri, value,
                   source, feed_id, valid_from, valid_to, 0 AS src_rank
            FROM ores_marketdata_market_observations_tbl
            WHERE tenant_id = $1 AND observation_datetime <= $2 AND valid_to = $3
            UNION ALL
            SELECT md5(c.series_id::text || t.coord::text || t.ts::text)::uuid,
                   c.tenant_id, c.party_id, c.series_id, t.ts, c.oresmd_uri, t.value::text,
                   NULL::text, t.feed_id, t.ts, $3::timestamptz, 1
            FROM coords c
            CROSS JOIN LATERAL (
                SELECT * FROM ores_marketdata_ticks_tbl x
                WHERE x.stream_id = c.stream_id AND x.coord = c.coord
                    AND x.ts <= $2::timestamptz
                ORDER BY x.ts DESC
                LIMIT 1
            ) t
        )
        SELECT DISTINCT ON (series_id, oresmd_uri)
            id, tenant_id, party_id, series_id, observation_datetime, oresmd_uri, value,
            source, feed_id, valid_from, valid_to
        FROM points
        ORDER BY series_id, oresmd_uri, observation_datetime DESC, src_rank
    )";

    const auto rows = query_builder::execute_parameterized_multi_column_query(
        ctx,
        sql,
        {tid, as_of_str, MAX_TIMESTAMP, pg_array(series_ids), pg_array(streams)},
        lg(),
        "reading as-of snapshot of every series");

    std::vector<observation_record> result;
    result.reserve(rows.size());
    for (const auto& row : rows)
        result.push_back(
            {market_observation_mapper::map(as_of_observation(row, 0, "read_as_of_all")),
             as_of_recorded_at(row, 0, "read_as_of_all")});

    // A current manual point owns its datum until cleared, as in the read for one series.
    std::map<std::pair<std::string, std::string>, std::size_t> position;
    for (std::size_t i = 0; i != result.size(); ++i)
        position[{boost::uuids::to_string(result[i].observation.series_id),
                  result[i].observation.oresmd_uri}] = i;
    for (auto& manual : read_manual_points_all(ctx, lg())) {
        if (manual.observation.observation_datetime > as_of_datetime)
            continue;
        const auto key = std::pair{boost::uuids::to_string(manual.observation.series_id),
                                   manual.observation.oresmd_uri};
        if (const auto it = position.find(key); it != position.end())
            result[it->second] = std::move(manual);
        else
            result.push_back(std::move(manual));
    }

    std::vector<domain::market_observation> observations;
    observations.reserve(result.size());
    for (auto& record : result)
        observations.push_back(std::move(record.observation));
    return observations;
}

std::vector<std::vector<domain::market_observation>>
read_as_of_buckets(context ctx,
                   const boost::uuids::uuid& series_id,
                   const std::chrono::system_clock::time_point& latest_boundary,
                   const std::chrono::seconds& bucket_size,
                   unsigned int bucket_count);

std::vector<std::vector<domain::market_observation>>
market_observation_repository::read_as_of_buckets(
    context ctx,
    const boost::uuids::uuid& series_id,
    const std::chrono::system_clock::time_point& latest_boundary,
    const std::chrono::seconds& bucket_size,
    unsigned int bucket_count) {
    using ores::platform::time::datetime;
    BOOST_LOG_SEV(lg(), debug) << "Reading " << bucket_count
                               << " as-of bucket snapshots for series: " << series_id << " every "
                               << bucket_size.count()
                               << "s ending at: " << datetime::to_iso8601_utc(latest_boundary);
    const auto tid = ctx.tenant_id().to_string();
    const auto sid = boost::uuids::to_string(series_id);
    const auto latest_str = datetime::to_iso8601_utc(latest_boundary);
    const auto bucket_seconds = std::to_string(bucket_size.count());
    const auto count_str = std::to_string(bucket_count);
    const auto stream = stream_of(ctx, series_id, may_read_ticks);

    // Bucket boundaries and the per-bucket as-of reduction both happen in the database, in
    // one statement: generate_series() produces the (up to bucket_count) boundaries ending
    // at latest_boundary, and the LATERAL subquery resolves each oresmd_uri to its own latest
    // observation at or before that boundary -- a per-bucket DISTINCT ON (oresmd_uri), driven
    // by observations_series_coordinate_datetime_idx (tenant_id, series_id, oresmd_uri,
    // observation_datetime desc), so each point is a skip-scan straight to its latest row
    // per bucket rather than a sort over the whole series. bucket_ordinal lets the caller
    // regroup rows by bucket without relying on floating-point/timestamp equality.
    static const std::string sql = R"(
        WITH boundaries AS (
            SELECT ordinality - 1 AS bucket_ordinal, boundary
            FROM generate_series(
                $3::timestamptz - ($5::int - 1) * ($4::int * interval '1 second'),
                $3::timestamptz,
                $4::int * interval '1 second'
            ) WITH ORDINALITY AS t(boundary, ordinality)
        ),
        )" + tick_coords_cte("$7", "$6") +
                                   R"(
        SELECT b.bucket_ordinal, o.id, o.tenant_id, o.party_id, o.series_id,
               o.observation_datetime, o.oresmd_uri, o.value, o.source, o.feed_id, o.valid_from, o.valid_to
        FROM boundaries b
        CROSS JOIN LATERAL (
            SELECT DISTINCT ON (u.oresmd_uri)
                u.id, u.tenant_id, u.party_id, u.series_id, u.observation_datetime,
                u.oresmd_uri, u.value, u.source, u.feed_id, u.valid_from, u.valid_to
            FROM (
                (SELECT DISTINCT ON (oresmd_uri)
                    id, tenant_id, party_id, series_id, observation_datetime, oresmd_uri, value,
                    source, feed_id, valid_from, valid_to, 0 AS src_rank
                FROM ores_marketdata_market_observations_tbl m
                WHERE m.tenant_id = $1 AND m.series_id = $2
                    AND m.observation_datetime <= b.boundary
                    AND m.valid_to = $6
                ORDER BY oresmd_uri, observation_datetime DESC)
                UNION ALL)" + latest_tick_points("b.boundary", "$6") +
                                   R"(
            ) u
            ORDER BY u.oresmd_uri, u.observation_datetime DESC, u.src_rank
        ) o
        ORDER BY b.bucket_ordinal, o.oresmd_uri
    )";

    const auto rows = query_builder::execute_parameterized_multi_column_query(
        ctx,
        sql,
        {tid, sid, latest_str, bucket_seconds, count_str, MAX_TIMESTAMP, stream},
        lg(),
        "reading as-of bucketed curve evolution");

    std::vector<std::vector<domain::market_observation>> result(bucket_count);
    for (const auto& row : rows) {
        const auto ordinal = as_of_bucket_ordinal(row, result.size());
        result[ordinal].push_back(
            market_observation_mapper::map(as_of_observation(row, 1, "read_as_of_buckets")));
    }

    // The manual overlay, once per bucket: a manual point owns its coordinate
    // from the bucket whose boundary is at or after the manual point's own
    // instant, and the buckets before it keep the fed value they showed. The
    // boundaries mirror the generate_series() above, where bucket_ordinal 0 is
    // the oldest.
    for (const auto& manual : read_manual_points(ctx, series_id, lg())) {
        for (unsigned int i = 0; i < bucket_count; ++i) {
            const auto offset = std::chrono::seconds(
                static_cast<long long>(bucket_count - 1 - i) * bucket_size.count());
            if (manual.observation.observation_datetime > latest_boundary - offset)
                continue;
            const auto it = std::ranges::find(
                result[i],
                manual.observation.oresmd_uri,
                &domain::market_observation::oresmd_uri);
            if (it != result[i].end())
                *it = manual.observation;
            else
                result[i].push_back(manual.observation);
        }
    }
    return result;
}

// One instant of a series: the points the series states at it.
struct series_instant {
    std::chrono::system_clock::time_point as_of;
    std::vector<domain::market_observation> points;
};

std::vector<series_instant>
read_instants(context ctx,
              const boost::uuids::uuid& series_id,
              const std::chrono::system_clock::time_point& from_datetime,
              const std::chrono::system_clock::time_point& to_datetime,
              const std::vector<std::chrono::system_clock::time_point>& instants,
              unsigned int max_instants);

std::vector<series_instant>
market_observation_repository::read_instants(
    context ctx,
    const boost::uuids::uuid& series_id,
    const std::chrono::system_clock::time_point& from_datetime,
    const std::chrono::system_clock::time_point& to_datetime,
    const std::vector<std::chrono::system_clock::time_point>& instants,
    unsigned int max_instants) {
    using ores::platform::time::datetime;
    BOOST_LOG_SEV(lg(), debug) << "Reading the instants of series: " << series_id << " from "
                               << datetime::to_iso8601_utc(from_datetime) << " to "
                               << datetime::to_iso8601_utc(to_datetime) << ", " << instants.size()
                               << " stated instant(s)";
    const auto tid = ctx.tenant_id().to_string();
    const auto sid = boost::uuids::to_string(series_id);
    const auto from_str = datetime::to_iso8601_utc(from_datetime);
    const auto to_str = datetime::to_iso8601_utc(to_datetime);
    const auto cap_str = std::to_string(max_instants + 1);
    const auto stream = stream_of(ctx, series_id, may_read_ticks);

    // The stated instants travel as one comma-separated string, because the
    // read takes text parameters. An empty one gives an empty array, which is
    // what the SQL reads as "no set was stated".
    std::string stated_str;
    for (const auto& instant : instants) {
        if (!stated_str.empty())
            stated_str += ',';
        stated_str += datetime::to_iso8601_utc(instant);
    }

    // The instants the read covers and their points come back together: the
    // boundaries are the caller's stated set when there is one and otherwise
    // the instants the series states in the range, and the points are the rows
    // at those instants and no others. The index is
    // observations_series_coordinate_datetime_idx (tenant_id, series_id,
    // oresmd_uri, observation_datetime desc), so the join by instant is served
    // by its leading columns rather than by a scan of the hypertable.
    //
    // The instant travels as microseconds since the epoch, because it is a
    // grouping key before it is a timestamp, and a text rendering parsed back
    // would group at another precision than the rows carry.
    static const std::string sql = R"(
        WITH stated AS (
            SELECT unnest(string_to_array($7, ',')::timestamptz[]) AS instant
        ),
        )" + tick_coords_cte("$8", "$6") +
                                   R"(,
        instants AS (
            SELECT instant FROM (
                SELECT instant FROM stated
                UNION
                SELECT DISTINCT observation_datetime
                FROM ores_marketdata_market_observations_tbl
                WHERE tenant_id = $1 AND series_id = $2
                    AND observation_datetime >= $3 AND observation_datetime <= $4
                    AND valid_to = $6
                    AND NOT EXISTS (SELECT 1 FROM stated)
                UNION
                SELECT DISTINCT ts
                FROM ores_marketdata_ticks_tbl
                WHERE stream_id = $8::uuid AND ts >= $3::timestamptz AND ts <= $4::timestamptz
                    AND NOT EXISTS (SELECT 1 FROM stated)
            ) u
            ORDER BY instant
            LIMIT $5::int
        ),
        points AS (
            SELECT i.instant, o.id, o.tenant_id, o.party_id, o.series_id, o.observation_datetime,
                   o.oresmd_uri, o.value, o.source, o.feed_id, o.valid_from, o.valid_to, 0 AS src_rank
            FROM instants i
            JOIN ores_marketdata_market_observations_tbl o
              ON o.tenant_id = $1 AND o.series_id = $2
             AND o.observation_datetime = i.instant AND o.valid_to = $6
            UNION ALL
            SELECT i.instant, md5(c.series_id::text || t.coord::text || t.ts::text)::uuid,
                   c.tenant_id, c.party_id, c.series_id, t.ts, c.oresmd_uri, t.value::text,
                   NULL::text, t.feed_id, t.ts, $6::timestamptz, 1
            FROM instants i
            JOIN ores_marketdata_ticks_tbl t ON t.stream_id = $8::uuid AND t.ts = i.instant
            JOIN coords c ON c.stream_id = t.stream_id AND c.coord = t.coord
        )
        SELECT DISTINCT ON (instant, oresmd_uri)
               (extract(epoch FROM instant) * 1000000)::bigint::text,
               id, tenant_id, party_id, series_id, observation_datetime,
               oresmd_uri, value, source, feed_id, valid_from, valid_to
        FROM points
        ORDER BY instant, oresmd_uri, src_rank
    )";

    const auto rows = query_builder::execute_parameterized_multi_column_query(
        ctx,
        sql,
        {tid, sid, from_str, to_str, cap_str, MAX_TIMESTAMP, stated_str, stream},
        lg(),
        "reading the instants of a series");

    std::vector<series_instant> result;
    for (const auto& row : rows) {
        if (row.empty() || !row[0])
            throw std::runtime_error("read_instants: a row carries no instant");
        const std::chrono::system_clock::time_point instant{
            std::chrono::microseconds(std::stoll(*row[0]))};
        if (result.empty() || result.back().as_of != instant) {
            if (result.size() >= max_instants)
                throw std::runtime_error("read_instants: the range holds more than " +
                                         std::to_string(max_instants) +
                                         " instants; ask for a narrower range");
            result.push_back({instant, {}});
        }
        result.back().points.push_back(
            market_observation_mapper::map(as_of_observation(row, 1, "read_instants")));
    }
    return result;
}

void write_manual_point(context ctx,
                        const boost::uuids::uuid& series_id,
                        const std::string& oresmd_uri,
                        std::chrono::system_clock::time_point observation_datetime,
                        const std::string& value,
                        const std::string& change_reason_code,
                        const std::string& change_commentary);

void market_observation_repository::write_manual_point(
    context ctx,
    const boost::uuids::uuid& series_id,
    const std::string& oresmd_uri,
    std::chrono::system_clock::time_point observation_datetime,
    const std::string& value,
    const std::string& change_reason_code,
    const std::string& change_commentary) {
    BOOST_LOG_SEV(lg(), debug) << "Writing manual point for series: " << series_id
                               << " coordinate: " << oresmd_uri;
    const auto party_id = ctx.party_id();
    if (!party_id)
        throw std::invalid_argument(
            "market_observation_repository::write_manual_point: the context names no party");

    // The actor names who keyed the point. A request context carries one; a
    // system-initiated write falls back to the service account, exactly as the
    // messaging stamp helper does.
    const auto actor = ctx.actor().empty() ? ctx.service_account() : ctx.actor();

    // One transaction writes the observation row and its annex: the read shows
    // a manual point only when both are present, so a failure in either leaves
    // neither behind.
    unit_of_work uow(ctx);
    const auto& txn_ctx = uow.ctx();

    observation_lineage_repository lineage_repo;
    const auto current = lineage_repo.read_latest_by_observation(
        txn_ctx, series_id, observation_datetime, oresmd_uri);

    domain::market_observation obs;
    obs.id = boost::uuids::random_generator()();
    obs.tenant_id = ctx.tenant_id();
    obs.party_id = *party_id;
    obs.series_id = series_id;
    obs.observation_datetime = observation_datetime;
    obs.oresmd_uri = oresmd_uri;
    obs.value = value;
    obs.source = "manual.operator";
    // A manual point is written past the insert's guard, which drops an automatic
    // write to a coordinate an operator owns, and this write is the operator's.
    series_shape_check::check(txn_ctx, std::vector<domain::market_observation>{obs});
    query_builder::execute_write_query(
        txn_ctx, market_observation_mapper::map(obs), lg(), "Inserting manual market observation.");

    domain::observation_lineage lineage;
    // An annex row already at this natural key -- a derivation's or an earlier
    // manual point's -- keeps its own id, so the insert trigger closes it and
    // writes the new generation instead of colliding with the current-row key.
    lineage.id = current ? current->id : boost::uuids::random_generator()();
    lineage.tenant_id = ctx.tenant_id();
    lineage.party_id = *party_id;
    lineage.series_id = series_id;
    lineage.observation_datetime = observation_datetime;
    lineage.oresmd_uri = oresmd_uri;
    lineage.point_source_kind = "manual";
    lineage.derivation_config_id = std::nullopt;
    lineage.source_as_of = std::nullopt;
    lineage.source_series_ids = "[]";
    lineage.modified_by = actor;
    lineage.performed_by = ctx.service_account();
    lineage.change_reason_code = change_reason_code;
    lineage.change_commentary = change_commentary;
    lineage_repo.write(txn_ctx, lineage);

    uow.commit();
}

void clear_manual_point(context ctx,
                        const boost::uuids::uuid& series_id,
                        const std::string& oresmd_uri,
                        std::chrono::system_clock::time_point observation_datetime);

void market_observation_repository::clear_manual_point(
    context ctx,
    const boost::uuids::uuid& series_id,
    const std::string& oresmd_uri,
    std::chrono::system_clock::time_point observation_datetime) {
    BOOST_LOG_SEV(lg(), debug) << "Clearing manual point for series: " << series_id
                               << " coordinate: " << oresmd_uri;

    // The read that finds the row and the close that withdraws it share one
    // transaction, so a point that moved on between them is not closed twice.
    unit_of_work uow(ctx);
    const auto& txn_ctx = uow.ctx();

    observation_lineage_repository lineage_repo;
    const auto current = lineage_repo.read_latest_by_observation(
        txn_ctx, series_id, observation_datetime, oresmd_uri);
    // Only a manual point is withdrawn: a derived point's annex is not this
    // operation's to close, and a coordinate with no annex is quoted.
    if (!current || current->point_source_kind != "manual")
        return;

    // Closing, not deleting: the delete rule keeps the row in the annex's
    // history, so a reader still sees what the over-key replaced.
    lineage_repo.remove(txn_ctx, boost::uuids::to_string(current->id));

    // The over-key's observation row goes with its annex. The delete rule keeps
    // it in the observation history, as it does the annex.
    std::vector<std::string> keyed_ids;
    for (const auto& obs : read_latest_for_series(txn_ctx, series_id)) {
        if (obs.source == "manual.operator" && obs.oresmd_uri == oresmd_uri &&
            obs.observation_datetime == observation_datetime)
            keyed_ids.push_back(boost::uuids::to_string(obs.id));
    }
    remove(txn_ctx, keyed_ids);
    uow.commit();
}

std::size_t clear_manual_points(context ctx,
                                const boost::uuids::uuid& series_id,
                                const std::string& oresmd_uri);

std::size_t market_observation_repository::clear_manual_points(
    context ctx, const boost::uuids::uuid& series_id, const std::string& oresmd_uri) {
    BOOST_LOG_SEV(lg(), debug) << "Clearing every manual point for series: " << series_id
                               << " coordinate: " << oresmd_uri;

    unit_of_work uow(ctx);
    const auto& txn_ctx = uow.ctx();

    observation_lineage_repository lineage_repo;
    std::vector<std::chrono::system_clock::time_point> instants;
    for (const auto& annex : lineage_repo.read_manual_for_series(txn_ctx, series_id)) {
        if (annex.oresmd_uri != oresmd_uri)
            continue;
        lineage_repo.remove(txn_ctx, boost::uuids::to_string(annex.id));
        if (std::ranges::find(instants, annex.observation_datetime) == instants.end())
            instants.push_back(annex.observation_datetime);
    }
    if (instants.empty())
        return 0;

    // The keyed observation rows go with their annex rows, so the value the feed
    // last wrote shows. The delete rule keeps them in the observation history.
    std::vector<std::string> keyed_ids;
    for (const auto& obs : read_latest_for_series(txn_ctx, series_id)) {
        if (obs.source == "manual.operator" && obs.oresmd_uri == oresmd_uri &&
            std::ranges::find(instants, obs.observation_datetime) != instants.end())
            keyed_ids.push_back(boost::uuids::to_string(obs.id));
    }
    remove(txn_ctx, keyed_ids);
    uow.commit();
    return instants.size();
}

#include "ores.marketdata.core/repository/manual_point_guard.hpp"
#include "ores.marketdata.core/repository/series_shape_check.hpp"

series_shape_check::check(ctx, std::vector<domain::market_observation>{v});
if (manual_point_guard::unshadowed(ctx, std::vector<domain::market_observation>{v}).empty())
    return;

series_shape_check::check(ctx, v);
if (const auto kept = manual_point_guard::unshadowed(ctx, v); kept.size() != v.size()) {
    if (!kept.empty())
        query_builder::execute_write_query(ctx,
                            market_observation_mapper::map(kept),
                            lg(),
                            "Inserting market observations into database.");
    return;
}