static USAGE: &str = r#"
Geocodes a location in CSV data against an updatable local copy of the Geonames cities index
and a local copy of the MaxMind GeoLite2 City database.

The Geonames cities index can be retrieved and updated using the `geocode index-*` subcommands.

The GeoLite2 City database will need to be MANUALLY downloaded from MaxMind. Though it is
free, you will need to create a MaxMind account to download the GeoIP2 Binary database (mmdb)
from https://www.maxmind.com/en/accounts/current/geoip/downloads.
Copy the GeoLite2-City.mmdb file to the ~/.qsv-cache/ directory or point to it using the
QSV_GEOIP2_FILENAME environment variable.

When you run the command for the first time, it will download a prebuilt Geonames cities
index from the qsv GitHub repo and use it going forward. You can operate on the local
index using the `geocode index-*` subcommands.

By default, the prebuilt index uses the Geonames Gazeteer cities15000.zip file using
English names. It contains cities with populations > 15,000 (about ~26k cities). 
See https://download.geonames.org/export/dump/ for more information.

It has twelve major subcommands:
 * suggest        - given a partial City name, return the closest City's location metadata
                    per the local Geonames cities index (Jaro-Winkler distance)
 * suggestnow     - same as suggest, but using a partial City name from the command line,
                    instead of CSV data.
 * reverse        - given a WGS-84 location coordinate, return the closest City's location
                    metadata per the local Geonames cities index.
                    (Euclidean distance - shortest distance "as the crow flies")
 * reversenow     - sames as reverse, but using a coordinate from the command line,
                    instead of CSV data.
 * countryinfo    - returns the country information for the ISO-3166 2-letter country code
                    (e.g. US, CA, MX, etc.)
 * countryinfonow - same as countryinfo, but using a country code from the command line,
                    instead of CSV data.
 * iplookup       - given an IP address or URL, return the closest City's location metadata
                    per the local Maxmind GeoLite2 City database.
 * iplookupnow    - same as iplookup, but using an IP address or URL from the command line,
                    instead of CSV data.
 * opencage       - ONLINE forward/reverse geocoding using the OpenCage API.
                    Forward-geocodes a free-form address, or reverse-geocodes a
                    "lat, long" coordinate. Requires an OpenCage API key.
 * opencagenow    - same as opencage, but using an address/coordinate from the
                    command line, instead of CSV data.
 * index-*        - operations to update the local Geonames cities index.
                    (index-check, index-update, index-load & index-reset)
 * cache-*        - operations to manage the persistent on-disk OpenCage result cache.
                    (cache-clear, cache-prune & cache-info)

SUGGEST
Suggest a Geonames city based on a partial city name. It returns the closest Geonames
city record based on the Jaro-Winkler distance between the partial city name and the
Geonames city name.

The geocoded information is formatted based on --formatstr, returning it in 
'%location' format (i.e. "(lat, long)") if not specified.

Use the --new-column option if you want to keep the location column, e.g.

Geocode file.csv city column and set the geocoded value to a new column named lat_long.

  $ qsv geocode suggest city --new-column lat_long file.csv

Limit suggestions to the US, Canada and Mexico.

  $ qsv geocode suggest city --country us,ca,mx file.csv

Limit suggestions to New York State and California, with matches in New York state
having higher priority as its listed first.

  $ qsv geocode suggest city --country us --admin1 "New York,US.CA" file.csv

If we use admin1 codes, we can omit --country as it will be inferred from the admin1 code prefix.

  $ qsv geocode suggest city --admin1 "US.NY,US.CA" file.csv

Geocode file.csv city column with --formatstr=%state and set the 
geocoded value a new column named state.

  $ qsv geocode suggest city --formatstr %state --new-column state file.csv

Use dynamic formatting to create a custom format.

  $ qsv geocode suggest city -f "{name}, {admin1}, {country} in {timezone}" file.csv

Using French place names. You'll need to rebuild the index with the --languages option first

  $ qsv geocode suggest city -f "{name}, {admin1}, {country} in {timezone}" -l fr file.csv

SUGGESTNOW
Accepts the same options as suggest, but does not require an input file.
Its default format is more verbose - "{name}, {admin1} {country}: {latitude}, {longitude}"

  $ qsv geocode suggestnow "New York"
  $ qsv geocode suggestnow --country US -f %cityrecord "Paris"
  $ qsv geocode suggestnow --admin1 "US:OH" "Athens"

REVERSE
Reverse geocode a WGS 84 coordinate to the nearest City. It returns the closest Geonames
city record based on the Euclidean distance between the coordinate and the nearest city.
It accepts "lat, long" or "(lat, long)" format.

The geocoded information is formatted based on --formatstr, returning it in
'%city-admin1' format if not specified, e.g.

Reverse geocode file.csv LatLong column. Set the geocoded value to a new column named City.

  $ qsv geocode reverse LatLong -c City file.csv

Reverse geocode file.csv LatLong column and set the geocoded value to a new column
named CityState, output to a file named file_with_citystate.csv.

  $ qsv geocode reverse LatLong -c CityState file.csv -o file_with_citystate.csv

The same as above, but get the timezone instead of the city and state.

  $ qsv geocode reverse LatLong -f %timezone -c tz file.csv -o file_with_tz.csv

REVERSENOW
Accepts the same options as reverse, but does not require an input file.

  $ qsv geocode reversenow "40.71427, -74.00597"
  $ qsv geocode reversenow --country US -f %cityrecord "40.71427, -74.00597"
  $ qsv geocode reversenow "(39.32924, -82.10126)"

COUNTRYINFO
Returns the country information for the specified ISO-3166 2-letter country code.

  $ qsv geocode countryinfo country_col data.csv
  $ qsv geocode countryinfo --formatstr "%json" country_col data.csv
  $ qsv geocode countryinfo -f "%continent" country_col data.csv
  $ qsv geocode countryinfo -f "{country_name} ({fips}) in {continent}" country_col data.csv

COUNTRYINFONOW
Accepts the same options as countryinfo, but does not require an input file.

  $ qsv geocode countryinfonow US
  $ qsv geocode countryinfonow --formatstr "%pretty-json" US
  $ qsv geocode countryinfonow -f "%continent" US
  $ qsv geocode countryinfonow -f "{country_name} ({fips}) in {continent}" US

IPLOOKUP
Given an IP address or URL, return the closest City's location metadata per the
local Geonames cities index.

  $ qsv geocode iplookup IP_col data.csv
  $ qsv geocode iplookup --formatstr "%json" IP_col data.csv
  $ qsv geocode iplookup -f "%cityrecord" IP_col data.csv

IPLOOKUPNOW
Accepts the same options as iplookup, but does not require an input file.

  $ qsv geocode iplookupnow 140.174.222.253
  $ qsv geocode iplookupnow https://amazon.com
  $ qsv geocode iplookupnow --formatstr "%json" 140.174.222.253
  $ qsv geocode iplookupnow -f "%cityrecord" 140.174.222.253

OPENCAGE
Online forward or reverse geocoding using the OpenCage Geocoding API
(https://opencagedata.com). Unlike the suggest/reverse subcommands which use the
local Geonames index, opencage geocodes real street addresses online.

Requires an OpenCage API key. Set it with --api-key or the QSV_OPENCAGE_API_KEY
environment variable (the --api-key flag takes precedence). Get a free key at
https://opencagedata.com/users/sign_up.

The <column> may contain either a free-form address (forward geocoding) or a
"lat, long" / "(lat, long)" WGS-84 coordinate (reverse geocoding). The mode is
auto-detected per row; pass --reverse to force reverse geocoding.

OpenCage's Terms of Service explicitly allow caching, so results are cached in a
persistent on-disk cache (see --cache-ttl & --no-cache). Re-runs and duplicate
queries do NOT re-hit the API. The free tier allows 2,500 requests/day at 1
request/second; rows are processed sequentially and rate-limited (see --rate-limit).

The --country option, if set, restricts results to the given ISO 3166-1 alpha-2
country code(s). The --timeout, --language, --invalid-result, --new-column, --rename
and --output options behave as they do for the other subcommands.

The --formatstr option supports these OpenCage-specific formats:
  * '%+' | '%formatted'   - the OpenCage formatted address (default)
  * '%lat-long'           - <latitude>, <longitude>
  * '%location'           - (<latitude>, <longitude>)
  * '%city'               - the city/town/village
  * '%state' | '%admin1'  - the state/province
  * '%county' | '%admin2' - the county
  * '%country'            - the ISO 3166-1 alpha-2 country code
  * '%country_name'       - the country name
  * '%postcode'           - the postal code
  * '%confidence'         - the OpenCage confidence score (0-10)
  * '%json'               - the first OpenCage result as JSON
  * '%pretty-json'        - the first OpenCage result as pretty JSON
Dynamic formatting is also supported, using dotted keys, e.g.
  "{components.city}, {components.country}" or "{annotations.timezone.name}".
Available keys: formatted, lat, lng, confidence, components.<name> and
annotations.<dotted.path>.

The special "%dyncols:" format is also supported, adding multiple columns to the
output CSV. Set --formatstr to "%dyncols:" followed by a comma-delimited list of
"{col_name:key}" pairs, where key is one of the dynamic keys above, e.g.
  "%dyncols: {city:components.city}, {tz:annotations.timezone.name}"
Like the other subcommands, "%dyncols:" cannot be combined with --new-column.

  $ qsv geocode opencage address --api-key YOURKEY file.csv
  $ qsv geocode opencage address --country us -f '%json' file.csv
  $ qsv geocode opencage coord_col --reverse -c city file.csv
  $ qsv geocode opencage address -f '{components.city}, {components.country}' file.csv
  $ qsv geocode opencage address -f '%dyncols: {city:components.city}, {pc:components.postcode}' file.csv

OPENCAGENOW
Accepts the same options as opencage, but does not require an input file.

  $ qsv geocode opencagenow --api-key YOURKEY "Brooklyn, NY"
  $ qsv geocode opencagenow "40.71427, -74.00597"
  $ qsv geocode opencagenow -f '%pretty-json' "Eiffel Tower, Paris"

INDEX-<operation>
Manage the local Geonames cities index used by the geocode command.

It has four operations:
 * check  - checks if the local Geonames index is up-to-date compared to the Geonames website.
            returns the index file's metadata JSON to stdout.
 * update - updates the local Geonames index with the latest changes from the Geonames website.
            use this command judiciously as it downloads about ~200mb of data from Geonames
            and rebuilds the index from scratch using the --languages option.
            If you don't need a language other than English, use the index-load subcommand instead
            as it's faster and will not download any data from Geonames.
 * reset  - resets the local Geonames index to the default prebuilt, English-only Geonames cities
            index (cities15000) - downloading it from the qsv GitHub repo for the current qsv version.
 * load   - load a Geonames cities index from a file, making it the default index going forward.
            If set to 15000, it will download the prebuilt English-only cities15000 Geonames
            index rkyv file from the qsv GitHub repo for the current qsv version.

Update the Geonames cities index with the latest changes.

  $ qsv geocode index-update

Rebuild the index using the latest Geonames data w/ English, French, German & Spanish place names

  $ qsv geocode index-update --languages en,fr,de,es

Load an alternative Geonames cities index from a file, making it the default index going forward.

  $ qsv geocode index-load my_geonames_index.rkyv

CACHE-<operation>
Manage the persistent on-disk OpenCage result cache used by the opencage subcommands.
This cache is separate from the Geonames cities index and is only populated by the
opencage/opencagenow subcommands. It lives in {cache-dir}/geocode-opencage_v1.

It has three operations:
 * clear  - wipe the entire OpenCage disk cache, removing all cached results.
 * prune  - delete cache entries older than the --older-than value. The value is either
            an absolute date/datetime (e.g. 2025-01-31, "2025-01-31 12:00:00") or a
            relative age with a unit suffix - s(econds), m(inutes), h(ours), d(ays) or
            w(eeks). e.g. 30d, 2w, 48h, 90m, 3600s.
 * info   - report the cache directory, entry count, on-disk size and the oldest/newest
            cached entry timestamps. Emits a JSON summary to stdout.

Wipe the entire OpenCage cache.

  $ qsv geocode cache-clear

Delete cached entries older than 30 days.

  $ qsv geocode cache-prune --older-than 30d

Delete cached entries created before a specific date.

  $ qsv geocode cache-prune --older-than 2025-01-01

Show cache statistics.

  $ qsv geocode cache-info

Examples:

# For US locations, you can retrieve the us_state_fips_code and us_county_fips_code fields of a US City
# to help with Census data enrichment.
qsv geocode suggest city_col --country US -f \
"%dyncols: {geocoded_city_col:name}, {state_col:admin1}, {county_col:admin2},  {state_fips_code:us_state_fips_code}, {county_fips_code:us_county_fips_code}"\
    input_data.csv -o output_data_with_fips.csv

# For US locations, you can reverse geocode the us_state_fips_code and us_county_fips_code fields of a WGS 84 coordinate
# to help with Census data enrichment. The coordinate can be in "lat, long" or "(lat, long)" format.
qsv geocode reverse wgs84_coordinate_col --country US -f \
"%dyncols: {geocoded_city_col:name}, {state_col:admin1}, {county_col:admin2},  {state_fips_code:us_state_fips_code}, {county_fips_code:us_county_fips_code}"\
    input_data.csv -o output_data_with_fips.csv

For more examples, see https://github.com/dathere/qsv/blob/master/tests/test_geocode.rs.
See also https://github.com/dathere/qsv/wiki/Geospatial#geocode

Usage:
qsv geocode suggest [--formatstr=<string>] [options] <column> [<input>]
qsv geocode suggestnow [options] <location>
qsv geocode reverse [--formatstr=<string>] [options] <column> [<input>]
qsv geocode reversenow [options] <location>
qsv geocode countryinfo [options] <column> [<input>]
qsv geocode countryinfonow [options] <location>
qsv geocode iplookup [options] <column> [<input>]
qsv geocode iplookupnow [options] <location>
qsv geocode opencage [--formatstr=<string>] [options] <column> [<input>]
qsv geocode opencagenow [options] <location>
qsv geocode index-load <index-file>
qsv geocode index-check
qsv geocode index-update [--languages=<lang>] [--cities-url=<url>] [--force] [--timeout=<seconds>]
qsv geocode index-reset
qsv geocode cache-clear [options]
qsv geocode cache-prune --older-than=<val> [options]
qsv geocode cache-info [options]
qsv geocode --help

geocode arguments:
        
    <input>                     The input file to read from. If not specified, reads from stdin.

    <column>                    The column to geocode. Used by suggest, reverse & countryinfo subcommands.
                                For suggest, it must be a column with a City string pattern.
                                For reverse, it must be a column using WGS 84 coordinates in
                                "lat, long" or "(lat, long)" format.
                                For countryinfo, it must be a column with a ISO 3166-1 alpha-2 country code.
                                For iplookup, it must be a column with an IP address or a URL.
                                For opencage, it may be a free-form address OR a WGS 84 coordinate.
                                Note that you can use column selector syntax to select the column, but only
                                the first column will be used. See `select --help` for more information.

    <location>                  The location to geocode for suggestnow, reversenow, countryinfonow and
                                iplookupnow subcommands.
                                  For suggestnow, its a City string pattern.
                                  For reversenow, it must be a WGS 84 coordinate.
                                  For countryinfonow, it must be a ISO 3166-1 alpha-2 code.
                                  For iplookupnow, it must be an IP address or a URL.
                                  For opencagenow, it must be an address OR a WGS 84 coordinate.

    <index-file>                The alternate geonames index file to use. It must be a .rkyv file.
                                For convenience, if this is set to 15000, it will download the prebuilt
                                English-only cities15000 Geonames index rkyv file from the qsv GitHub repo
                                for the current qsv version and use it. Only used by the index-load subcommand.

geocode options:
    -c, --new-column <name>     Put the transformed values in a new column instead. Not valid when
                                using the '%dyncols:' --formatstr option.
    -r, --rename <name>         New name for the transformed column.
    --country <country_list>    The comma-delimited, case-insensitive list of countries to filter for.
                                Country is specified as a ISO 3166-1 alpha-2 (two-letter) country code.
                                https://en.wikipedia.org/wiki/ISO_3166-2

                                It is the topmost priority filter, and will be applied first. If multiple
                                countries are specified, they are matched in priority order.
                                
                                For suggest, this will limit the search to the specified countries.

                                For reverse, this ensures that the returned city is in the specified
                                countries (especially when geocoding coordinates near country borders).
                                If the coordinate is outside the specified countries, the returned city
                                will be the closest city as the crow flies in the specified countries.

                                SUGGEST only options:
    --min-score <score>         The minimum Jaro-Winkler distance score.
                                [default: 0.8]
    --admin1 <admin1_list>      The comma-delimited, case-insensitive list of admin1s to filter for.
    
                                If all uppercase, it will be treated as an admin1 code (e.g. US.NY, JP.40, CN.23).
                                Otherwise, it will be treated as an admin1 name (e.g New York, Tokyo, Shanghai).

                                Requires the --country option. However, if all admin1 codes have the same
                                prefix (e.g. US.TX, US.NJ, US.CA), the country can be inferred from the
                                admin1 code (in this example - US), and the --country option is not required.

                                If specifying multiple admin1 filters, you can mix admin1 codes and names,
                                and they are matched in priority order.

                                Matches are made using a starts_with() comparison (i.e. "US" will match "US.NY",
                                "US.NJ", etc. for admin1 code. "New" will match "New York", "New Jersey",
                                "Newfoundland", etc. for admin1 name.)

                                admin1 is the second priority filter, and will be applied after country filters.
                                See https://download.geonames.org/export/dump/admin1CodesASCII.txt for
                                recognized admin1 codes/names.

                                REVERSE only option:
    -k, --k_weight <weight>     Use population-weighted distance for reverse subcommand.
                                (i.e. nearest.distance - k * city.population)
                                Larger values will favor more populated cities.
                                If not set (default), the population is not used and the
                                nearest city is returned.

                                OPENCAGE only options:
    --api-key <key>             The OpenCage API key for the opencage/opencagenow subcommands.
                                If set, it takes precedence over the QSV_OPENCAGE_API_KEY
                                environment variable. Get a free key at
                                https://opencagedata.com/users/sign_up.
    --rate-limit <qps>          Maximum number of OpenCage API requests per second.
                                The free tier allows 1 request/second (2,500/day).
                                [default: 1]
    --reverse                   Force reverse geocoding for opencage/opencagenow (treat the
                                query as a "lat, long" WGS-84 coordinate). If not set, forward
                                and reverse mode is auto-detected per row.
    --no-annotations            Omit OpenCage annotations (timezone, currency, etc.) from the
                                result and from %json output.
    --cache-ttl <seconds>       Time-to-live for the persistent on-disk OpenCage result cache.
                                A value of 0 disables time-based expiration (entries are
                                cached indefinitely). Use --no-cache to disable caching entirely.
                                [default: 1209600]
    --no-cache                  Disable the persistent on-disk OpenCage cache. Duplicate
                                queries within a run are still de-duplicated.

    -f, --formatstr=<string>    The place format to use. It has three options:
                                1. Use one of the predefined formats.
                                2. Use dynamic formatting to create a custom format.
                                3. Use the special format "%dyncols:" to dynamically add multiple
                                   columns to the output CSV using fields from a geocode result.
    
                                PREDEFINED FORMATS:
                                  * '%city-state' - e.g. Brooklyn, New York
                                  * '%city-country' - Brooklyn, US
                                  * '%city-state-country' | '%city-admin1-country' - Brooklyn, New York US
                                  * '%city-county-state' | '%city-admin2-admin1' - Brooklyn, Kings County, New York
                                  * '%city' - Brooklyn
                                  * '%state' | '%admin1' - New York
                                  * '%county' | '%admin2' - Kings County
                                  * '%country' - US
                                  * '%country_name' - United States
                                  * '%cityrecord' - returns the full city record as a string
                                  * '%admin1record' - returns the full admin1 record as a string
                                  * '%admin2record' - returns the full admin2 record as a string
                                  * '%lat-long' - <latitude>, <longitude>
                                  * '%location' - (<latitude>, <longitude>)
                                  * '%id' - the Geonames ID
                                  * '%capital' - the capital
                                  * '%continent' - the continent (only valid for countryinfo subcommand)
                                  * '%population' - the population
                                  * '%timezone' - the timezone
                                  * '%json' - the full city record as JSON
                                  * '%pretty-json' - the full city record as pretty JSON
                                  * '%+' - use the subcommand's default format. 
                                           suggest - '%location'
                                           suggestnow - '{name}, {admin1} {country}: {latitude}, {longitude}'
                                           reverse & reversenow - '%city-admin1-country'
                                           countryinfo - '%country_name'
                                           iplookup - '%cityrecord'
                                           iplookupnow - '{name}, {admin1} {country}: {latitude}, {longitude}'

                                
                                If an invalid format is specified, it will be treated as '%+'.

                                Note that when using the JSON predefined formats with the now subcommands,
                                the output will be valid JSON, as the "Location" header will be omitted.

                                DYNAMIC FORMATTING:
                                Alternatively, you can use dynamic formatting to create a custom format.
                                To do so, set the --formatstr to a dynfmt template, enclosing field names
                                in curly braces.
                                The following ten cityrecord fields are available:
                                  id, name, latitude, longitude, country, admin1, admin2, capital,
                                  timezone, population

                                Fifteen additional countryinfo field are also available:
                                  iso3, fips, area, country_population, continent, tld, currency_code,
                                  currency_name, phone, postal_code_format, postal_code_regex, languages,
                                  country_geonameid, neighbours, equivalent_fips_code

                                For US places, two additional fields are available:
                                  us_county_fips_code and us_state_fips_code
                                    
                                  e.g. "City: {name}, State: {admin1}, Country: {country} {continent} - {languages}"

                                If an invalid template is specified, "Invalid dynfmt template" is returned.

                                Both predefined and dynamic formatting are cached. Subsequent calls
                                with the same result will be faster as it will use the cached result instead
                                of searching the Geonames index.

                                DYNAMIC COLUMNS ("%dyncols:") FORMATTING:
                                Finally, you can use the special format "%dyncols:" to dynamically add multiple
                                columns to the output CSV using fields from a geocode result.
                                To do so, set --formatstr to "%dyncols:" followed by a comma-delimited list
                                of key:value pairs enclosed in curly braces.
                                The key is the desired column name and the value is one of the same fields
                                available for dynamic formatting.

                                 e.g. "%dyncols: {city_col:name}, {state_col:admin1}, {county_col:admin2}"

                                will add three columns to the output CSV named city_col, state_col & county_col.

                                Note that using "%dyncols:" will cause the command to geocode EACH row without
                                using the cache, so it will be slower than predefined or dynamic formatting.
                                Also, countryinfo and countryinfonow subcommands currently do not support "%dyncols:".
                                [default: %+]
    -l, --language <lang>       The language to use when geocoding. The language is specified as a ISO 639-1 code.
                                Note that the Geonames index must have been built with the specified language
                                using the `index-update` subcommand with the --languages option.
                                If the language is not available, the first language in the index is used.
                                [default: en]

    --invalid-result <string>   The string to return when the geocode result is empty/invalid.
                                If not set, the original value is used.
    -j, --jobs <arg>            The number of jobs to run in parallel.
                                When not set, the number of jobs is set to the number of CPUs detected.
    -b, --batch <size>          The number of rows per batch to load into memory, before running in parallel.
                                Set to 0 to load all rows in one batch.
                                [default: 50000]
    --timeout <seconds>         Timeout for downloading Geonames cities index.
                                [default: 120]
    --cache-dir <dir>           The directory to use for caching the Geonames cities index
                                and the persistent on-disk OpenCage result cache.
                                If the directory does not exist, qsv will attempt to create it.
                                If the QSV_CACHE_DIR envvar is set, it will be used instead.
                                [default: ~/.qsv-cache]

                                CACHE-PRUNE only option:
    --older-than <val>          Delete OpenCage cache entries older than this value.
                                Accepts an absolute date/datetime (e.g. 2025-01-31) or a
                                relative age with a unit suffix (s/m/h/d/w = seconds,
                                minutes, hours, days or weeks; e.g. 30d, 2w, 48h).
                                Required by the cache-prune subcommand.

                                INDEX-UPDATE only options:
    --languages <lang-list>     The comma-delimited, case-insensitive list of languages to use when building
                                the Geonames cities index.
                                The languages are specified as a comma-separated list of ISO 639-2 codes.
                                See https://download.geonames.org/export/dump/iso-languagecodes.txt to look up codes
                                and https://download.geonames.org/export/dump/alternatenames/ for the supported
                                language files. 253 languages are currently supported.
                                [default: en]
    --cities-url <url>          The URL to download the Geonames cities file from. There are several
                                available at https://download.geonames.org/export/dump/.
                                  cities500.zip   - cities with populations > 500; ~200k cities, 56mb
                                  cities1000.zip  - population > 1000; ~140k cities, 44mb
                                  cities5000.zip  - population > 5000; ~53k cities, 21mb
                                  cities15000.zip - population > 15000; ~26k cities, 13mb
                                Note that the more cities are included, the larger the local index file will be,
                                lookup times will be slower, and the search results will be different.
                                For convenience, if this is set to 500, 1000, 5000 or 15000, it will be 
                                converted to a geonames cities URL.
                                [default: https://download.geonames.org/export/dump/cities15000.zip]
    --force                     Force update the Geonames cities index. If not set, qsv will check if there
                                are updates available at Geonames.org before updating the index.

Common options:
    -h, --help                  Display this message
    -o, --output <file>         Write output to <file> instead of stdout.
    -d, --delimiter <arg>       The field delimiter for reading CSV data.
                                Must be a single character. (default: ,)
    -p, --progressbar           Show progress bars. Will also show the cache hit rate upon completion.
                                Not valid for stdin.
"#;

use std::{
    collections::HashMap,
    fs,
    net::{IpAddr, Ipv4Addr},
    num::NonZeroU32,
    path::{Path, PathBuf},
    time::{Duration, SystemTime},
};

use cached::{
    ConcurrentCacheBase, ConcurrentCached, RedbCache,
    macros::{cached, concurrent_cached},
};
use dynfmt2::Format;
use geosuggest_core::{
    Engine, EngineData,
    index::{ArchivedCitiesRecord as CitiesRecord, ArchivedCountryRecord as CountryRecord},
    storage,
};
use geosuggest_utils::{IndexUpdater, IndexUpdaterSettings, SourceItem};
use governor::{DefaultDirectRateLimiter, Quota, RateLimiter};
use indicatif::{HumanBytes, ProgressBar, ProgressDrawTarget};
use log::info;
use phf::phf_map;
use rayon::{
    iter::{IndexedParallelIterator, ParallelIterator},
    prelude::IntoParallelRefIterator,
};
use regex::Regex;
use serde::Deserialize;
use serde_json::json;
use tempfile::tempdir;
use url::Url;
use util::expand_tilde;
use uuid::Uuid;

use crate::{
    CliResult,
    clitypes::CliError,
    config::{Config, Delimiter},
    regex_oncelock,
    select::SelectColumns,
    util,
    util::replace_column_value,
};

// Cached regex patterns used throughout the geocode module
// Using module-level statics for better performance
static ADMIN1_CODE_REGEX: fn() -> &'static Regex = || regex_oncelock!(r"^[A-Z]{2}\.[A-Z0-9]{1,8}$");

static LOCATION_REGEX: fn() -> &'static Regex =
    || regex_oncelock!(r"(?-u)([+-]?(?:\d+\.?\d*|\.\d+)),\s*([+-]?(?:\d+\.?\d*|\.\d+))");

static FORMATSTR_REGEX: fn() -> &'static Regex = || regex_oncelock!(r"\{(?P<key>\w+)\}");

// like FORMATSTR_REGEX, but also matches dotted keys (e.g. {components.city},
// {annotations.timezone.name}) used by the opencage subcommands' dynamic formatting
static OPENCAGE_FORMATSTR_REGEX: fn() -> &'static Regex = || regex_oncelock!(r"\{([\w.-]+)\}");

// matches a relative age for the cache-prune --older-than option, e.g. 30d, 2w,
// 48h, 90m, 3600s (case-insensitive). Group 1 is the count, group 2 is the unit.
static RELATIVE_AGE_REGEX: fn() -> &'static Regex = || regex_oncelock!(r"(?i)^(\d+)\s*([smhdw])$");

#[derive(Deserialize)]
struct Args {
    arg_column:          String,
    arg_location:        String,
    cmd_suggest:         bool,
    cmd_suggestnow:      bool,
    cmd_reverse:         bool,
    cmd_reversenow:      bool,
    cmd_countryinfo:     bool,
    cmd_countryinfonow:  bool,
    cmd_iplookup:        bool,
    cmd_iplookupnow:     bool,
    cmd_opencage:        bool,
    cmd_opencagenow:     bool,
    cmd_index_check:     bool,
    cmd_index_update:    bool,
    cmd_index_load:      bool,
    cmd_index_reset:     bool,
    cmd_cache_clear:     bool,
    cmd_cache_prune:     bool,
    cmd_cache_info:      bool,
    arg_input:           Option<String>,
    arg_index_file:      Option<String>,
    flag_rename:         Option<String>,
    flag_country:        Option<String>,
    flag_min_score:      Option<f32>,
    flag_admin1:         Option<String>,
    flag_k_weight:       Option<f32>,
    flag_formatstr:      String,
    flag_language:       String,
    flag_invalid_result: Option<String>,
    flag_batch:          usize,
    flag_timeout:        u16,
    flag_cache_dir:      String,
    flag_languages:      String,
    flag_cities_url:     String,
    flag_force:          bool,
    flag_jobs:           Option<usize>,
    flag_new_column:     Option<String>,
    flag_output:         Option<String>,
    flag_delimiter:      Option<Delimiter>,
    flag_progressbar:    bool,
    flag_api_key:        Option<String>,
    flag_rate_limit:     u32,
    flag_reverse:        bool,
    flag_no_annotations: bool,
    flag_cache_ttl:      u64,
    flag_no_cache:       bool,
    flag_older_than:     Option<String>,
}

#[derive(Clone, Debug)]
struct Admin1Filter {
    admin1_string: String,
    is_code:       bool,
}

#[derive(Clone)]
struct NamesLang {
    cityname:    String,
    admin1name:  String,
    admin2name:  String,
    countryname: String,
}

/// `OpenCage` Geocoding API response. Only the fields qsv uses are deserialized.
/// See <https://opencagedata.com/api> for the full schema.
#[derive(Deserialize)]
struct OpencageResponse {
    #[serde(default)]
    results: Vec<OpencageResult>,
    #[serde(default)]
    status:  OpencageStatus,
}

#[derive(Deserialize, Default)]
struct OpencageStatus {
    #[serde(default)]
    code: u16,
}

#[derive(Deserialize)]
struct OpencageResult {
    #[serde(default)]
    formatted:   String,
    #[serde(default)]
    confidence:  u8,
    #[serde(default)]
    geometry:    OpencageGeometry,
    #[serde(default)]
    components:  HashMap<String, serde_json::Value>,
    #[serde(default)]
    annotations: serde_json::Value,
}

#[derive(Deserialize, Default)]
struct OpencageGeometry {
    #[serde(default)]
    lat: f64,
    #[serde(default)]
    lng: f64,
}

/// Error from an `OpenCage` lookup. `Fatal` aborts the whole run (bad API key, quota
/// exceeded); `Transient` is a per-row failure (network error, 5xx, rate-limited)
/// that is handled via --invalid-result.
enum OcError {
    Fatal(String),
    Transient(String),
}

static QSV_VERSION: &str = env!("CARGO_PKG_VERSION");
static DEFAULT_GEOCODE_INDEX_FILENAME: &str =
    concat!("qsv-", env!("CARGO_PKG_VERSION"), "-geocode-index.rkyv");
static GEOIP2_FILENAME: &str = "GeoLite2-City.mmdb";

static DEFAULT_CITIES_NAMES_URL: &str =
    "https://download.geonames.org/export/dump/alternateNamesV2.zip";
static DEFAULT_CITIES_NAMES_FILENAME: &str = "alternateNamesV2.txt";
static DEFAULT_COUNTRY_INFO_URL: &str = "https://download.geonames.org/export/dump/countryInfo.txt";
static DEFAULT_ADMIN1_CODES_URL: &str =
    "https://download.geonames.org/export/dump/admin1CodesASCII.txt";
static DEFAULT_ADMIN2_CODES_URL: &str = "https://download.geonames.org/export/dump/admin2Codes.txt";

// we use a compile time static perfect hash map for US state FIPS codes
static US_STATES_FIPS_CODES: phf::Map<&'static str, &'static str> = phf_map! {
    "AK" => "02",
    "AL" => "01",
    "AR" => "05",
    "AZ" => "04",
    "CA" => "06",
    "CO" => "08",
    "CT" => "09",
    "DC" => "11",
    "DE" => "10",
    "FL" => "12",
    "GA" => "13",
    "HI" => "15",
    "IA" => "19",
    "ID" => "16",
    "IL" => "17",
    "IN" => "18",
    "KS" => "20",
    "KY" => "21",
    "LA" => "22",
    "MA" => "25",
    "MD" => "24",
    "ME" => "23",
    "MI" => "26",
    "MN" => "27",
    "MO" => "29",
    "MS" => "28",
    "MT" => "30",
    "NC" => "37",
    "ND" => "38",
    "NE" => "31",
    "NH" => "33",
    "NJ" => "34",
    "NM" => "35",
    "NV" => "32",
    "NY" => "36",
    "OH" => "39",
    "OK" => "40",
    "OR" => "41",
    "PA" => "42",
    "RI" => "44",
    "SC" => "45",
    "SD" => "46",
    "TN" => "47",
    "TX" => "48",
    "UT" => "49",
    "VT" => "50",
    "VA" => "51",
    "WA" => "53",
    "WI" => "55",
    "WV" => "54",
    "WY" => "56",
    // the following are territories
    // and are not included in the default index
    // leaving them here for reference
    // "AS" => "60",
    // "GU" => "66",
    // "MP" => "69",
    // "PR" => "72",
    // "UM" => "74",
    // "VI" => "78",
};

static INVALID_DYNFMT: &str = "Invalid dynfmt template.";
static INVALID_COUNTRY_CODE: &str = "Invalid country code.";

// when suggesting with --admin1, how many suggestions to fetch from the engine
// before filtering by admin1
static SUGGEST_ADMIN1_LIMIT: usize = 10;

// valid column values for %dyncols
// when adding new columns, make sure to maintain the sort order
// otherwise, the dyncols check will fail as it uses binary search
static SORTED_VALID_DYNCOLS: [&str; 28] = [
    "admin1",
    "admin2",
    "area",
    "capital",
    "continent",
    "country",
    "country_geonameid",
    "country_name",
    "country_population",
    "currency_code",
    "currency_name",
    "equivalent_fips_code",
    "fips",
    "id",
    "iso3",
    "languages",
    "latitude",
    "longitude",
    "name",
    "neighbours",
    "phone",
    "population",
    "postal_code_format",
    "postal_code_regex",
    "timezone",
    "tld",
    "us_county_fips_code",
    "us_state_fips_code",
];

// dyncols populated sentinel value
static DYNCOLS_POPULATED: &str = "_POPULATED";

// valid subcommands
#[derive(Clone, Copy, PartialEq)]
enum GeocodeSubCmd {
    Suggest,
    SuggestNow,
    Reverse,
    ReverseNow,
    CountryInfo,
    CountryInfoNow,
    Iplookup,
    IplookupNow,
    Opencage,
    OpencageNow,
    IndexCheck,
    IndexUpdate,
    IndexLoad,
    IndexReset,
    CacheClear,
    CachePrune,
    CacheInfo,
}

pub fn run(argv: &[&str]) -> CliResult<()> {
    let mut args: Args = util::get_args(USAGE, argv)?;

    if args.flag_new_column.is_some() && args.flag_rename.is_some() {
        return fail_incorrectusage_clierror!(
            "Cannot use --new-column and --rename at the same time."
        );
    }

    if args.flag_new_column.is_some() && args.flag_formatstr.starts_with("%dyncols:") {
        return fail_incorrectusage_clierror!(
            "Cannot use --new-column with the '%dyncols:' --formatstr option."
        );
    }

    // validate & resolve options for the opencage/opencagenow subcommands
    if args.cmd_opencage || args.cmd_opencagenow {
        // resolve the API key: the --api-key flag takes precedence over the
        // QSV_OPENCAGE_API_KEY environment variable
        let api_key = args
            .flag_api_key
            .clone()
            .filter(|k| !k.trim().is_empty())
            .or_else(|| {
                std::env::var("QSV_OPENCAGE_API_KEY")
                    .ok()
                    .filter(|k| !k.trim().is_empty())
            });
        let Some(api_key) = api_key else {
            return fail_incorrectusage_clierror!(
                "OpenCage geocoding requires an API key. Set it with --api-key or the \
                 QSV_OPENCAGE_API_KEY environment variable. Get a free key at \
                 https://opencagedata.com/users/sign_up"
            );
        };
        args.flag_api_key = Some(api_key);

        if args.flag_admin1.is_some() {
            return fail_incorrectusage_clierror!(
                "The --admin1 filter is not supported by the opencage subcommands."
            );
        }
        if args.flag_rate_limit == 0 {
            return fail_incorrectusage_clierror!("--rate-limit must be >= 1.");
        }
    }

    // if args.flag_cities_url is a number and is 500, 1000, 5000 or 15000,
    // its a geonames cities file ID and convert it to a URL
    // we do this as a convenience shortcut for users
    if args.flag_cities_url.parse::<u16>().is_ok() {
        let cities_id = args.flag_cities_url;
        // ensure its a valid cities_id - 500, 1000, 5000 or 15000
        if cities_id != "500" && cities_id != "1000" && cities_id != "5000" && cities_id != "15000"
        {
            return fail_incorrectusage_clierror!(
                "Invalid --cities-url: {cities_id} - must be one of 500, 1000, 5000 or 15000"
            );
        }
        args.flag_cities_url =
            format!("https://download.geonames.org/export/dump/cities{cities_id}.zip");
    }

    if let Err(err) = Url::parse(&args.flag_cities_url) {
        return fail_incorrectusage_clierror!(
            "Invalid --cities-url: {url} - {err}",
            url = args.flag_cities_url,
            err = err
        );
    }

    // --older-than is only valid with cache-prune. Validate it eagerly so a bad
    // value errors regardless of whether an OpenCage cache exists yet.
    if args.flag_older_than.is_some() && !args.cmd_cache_prune {
        return fail_incorrectusage_clierror!(
            "--older-than is only valid with the `cache-prune` subcommand."
        );
    }
    if args.cmd_cache_prune
        && let Some(older_than) = args.flag_older_than.as_deref()
    {
        resolve_older_than(older_than)?;
    }

    // we need to use tokio runtime as geosuggest uses async
    let rt = tokio::runtime::Runtime::new()?;
    rt.block_on(geocode_main(args))?;

    Ok(())
}

// main async geocode function that does the actual work
#[allow(clippy::future_not_send)]
async fn geocode_main(args: Args) -> CliResult<()> {
    let mut index_cmd = false;
    let mut now_cmd = false;
    let mut iplookup_cmd = false;
    let geocode_cmd = if args.cmd_suggest {
        GeocodeSubCmd::Suggest
    } else if args.cmd_reverse {
        GeocodeSubCmd::Reverse
    } else if args.cmd_countryinfo {
        GeocodeSubCmd::CountryInfo
    } else if args.cmd_suggestnow {
        now_cmd = true;
        GeocodeSubCmd::SuggestNow
    } else if args.cmd_reversenow {
        now_cmd = true;
        GeocodeSubCmd::ReverseNow
    } else if args.cmd_countryinfonow {
        now_cmd = true;
        GeocodeSubCmd::CountryInfoNow
    } else if args.cmd_iplookup {
        iplookup_cmd = true;
        GeocodeSubCmd::Iplookup
    } else if args.cmd_iplookupnow {
        now_cmd = true;
        iplookup_cmd = true;
        GeocodeSubCmd::IplookupNow
    } else if args.cmd_opencage {
        GeocodeSubCmd::Opencage
    } else if args.cmd_opencagenow {
        now_cmd = true;
        GeocodeSubCmd::OpencageNow
    } else if args.cmd_index_check {
        index_cmd = true;
        GeocodeSubCmd::IndexCheck
    } else if args.cmd_index_update {
        index_cmd = true;
        GeocodeSubCmd::IndexUpdate
    } else if args.cmd_index_load {
        index_cmd = true;
        GeocodeSubCmd::IndexLoad
    } else if args.cmd_index_reset {
        index_cmd = true;
        GeocodeSubCmd::IndexReset
    } else if args.cmd_cache_clear {
        GeocodeSubCmd::CacheClear
    } else if args.cmd_cache_prune {
        GeocodeSubCmd::CachePrune
    } else if args.cmd_cache_info {
        GeocodeSubCmd::CacheInfo
    } else {
        // should not happen as docopt won't allow it
        unreachable!();
    };

    // setup cache directory
    let geocode_cache_dir = resolve_geocode_cache_dir(&args.flag_cache_dir)?;

    info!("Using cache directory: {}", geocode_cache_dir.display());

    // OpenCage subcommands geocode online via the OpenCage API and do not use the
    // local Geonames index, so dispatch before any index handling/loading.
    if matches!(
        geocode_cmd,
        GeocodeSubCmd::Opencage | GeocodeSubCmd::OpencageNow
    ) {
        return run_opencage(args, geocode_cmd, &geocode_cache_dir).await;
    }

    // cache-* subcommands manage the persistent on-disk OpenCage result cache.
    // They do not use the Geonames index, so dispatch before any index handling.
    if matches!(
        geocode_cmd,
        GeocodeSubCmd::CacheClear | GeocodeSubCmd::CachePrune | GeocodeSubCmd::CacheInfo
    ) {
        return run_cache_mgmt(&args, geocode_cmd, &geocode_cache_dir);
    }

    let geocode_index_filename = std::env::var("QSV_GEOCODE_INDEX_FILENAME")
        .unwrap_or_else(|_| DEFAULT_GEOCODE_INDEX_FILENAME.to_string());
    let active_geocode_index_file =
        format!("{}/{}", geocode_cache_dir.display(), geocode_index_filename);
    let geocode_index_file = args
        .arg_index_file
        .clone()
        .unwrap_or_else(|| active_geocode_index_file.clone());
    let geoip2_filename = std::env::var("QSV_GEOIP2_FILENAME")
        .unwrap_or_else(|_| format!("{}/{}", geocode_cache_dir.display(), GEOIP2_FILENAME));

    // create a TempDir for the one record CSV we're creating if we're doing a Now command
    // we're doing this at this scope so the TempDir is automatically dropped after we're done
    let tempdir = tempfile::Builder::new().prefix("qsv-geocode").tempdir()?;

    // we're doing a SuggestNow, ReverseNow or CountryInfoNow - create a one record CSV in tempdir
    // with one column named "Location" and the passed location value and use it as the input
    let input = if now_cmd {
        let tempdir_path = tempdir.path().to_string_lossy().to_string();
        let temp_csv_path = format!("{}/{}.csv", tempdir_path, Uuid::new_v4());
        let temp_csv_path = Path::new(&temp_csv_path);
        let mut temp_csv_wtr = csv::WriterBuilder::new().from_path(temp_csv_path)?;
        temp_csv_wtr.write_record(["Location"])?;
        temp_csv_wtr.write_record([&args.arg_location])?;
        temp_csv_wtr.flush()?;
        Some(temp_csv_path.to_string_lossy().to_string())
    } else {
        args.arg_input
    };

    let rconfig = Config::new(input.as_ref())
        .delimiter(args.flag_delimiter)
        .select(SelectColumns::parse(&args.arg_column)?);

    #[cfg(feature = "datapusher_plus")]
    let show_progress = false;

    // prep progress bar
    #[cfg(any(feature = "feature_capable", feature = "lite"))]
    let show_progress =
        (args.flag_progressbar || util::get_envvar_flag("QSV_PROGRESSBAR")) && !rconfig.is_stdin();

    let progress = ProgressBar::with_draw_target(None, ProgressDrawTarget::stderr_with_hz(5));
    if show_progress {
        util::prep_progress(&progress, util::count_rows(&rconfig)?);
    } else {
        progress.set_draw_target(ProgressDrawTarget::hidden());
    }

    if index_cmd {
        // cities_filename is derived from the cities_url
        // the filename is the last component of the URL with a .txt extension
        // e.g. https://download.geonames.org/export/dump/cities15000.zip -> cities15000.txt
        let cities_filename = args
            .flag_cities_url
            .split('/')
            .next_back()
            .unwrap_or("cities15000.zip")
            .replace(".zip", ".txt");

        // setup languages
        let languages_vec: Vec<&str> = args.flag_languages.split(',').map(AsRef::as_ref).collect();

        info!("geocode_index_file: {geocode_index_file} Languages: {languages_vec:?}");

        let indexupdater_settings = IndexUpdaterSettings {
            http_timeout_ms:  util::timeout_secs(args.flag_timeout)? * 1000,
            cities:           SourceItem {
                url:      &args.flag_cities_url,
                filename: &cities_filename,
            },
            names:            Some(SourceItem {
                url:      DEFAULT_CITIES_NAMES_URL,
                filename: DEFAULT_CITIES_NAMES_FILENAME,
            }),
            countries_url:    Some(DEFAULT_COUNTRY_INFO_URL),
            admin1_codes_url: Some(DEFAULT_ADMIN1_CODES_URL),
            admin2_codes_url: Some(DEFAULT_ADMIN2_CODES_URL),
            filter_languages: languages_vec.clone(),
        };

        let updater = IndexUpdater::new(indexupdater_settings.clone())
            .map_err(|_| CliError::Other("Error initializing IndexUpdater".to_string()))?;

        let index_storage = storage::Storage::new();

        match geocode_cmd {
            // check if Geoname index needs to be updated from the Geonames website
            // also returns the index file metadata as JSON
            GeocodeSubCmd::IndexCheck => {
                winfo!("Checking main Geonames website for updates...");
                check_index_file(&geocode_index_file)?;

                let metadata = index_storage
                    .read_metadata(geocode_index_file)
                    .map_err(|e| format!("index-check error: {e}"))?;

                let index_metadata_json = match simd_json::to_string_pretty(&metadata) {
                    Ok(json) => json,
                    Err(e) => {
                        let json_error = json!({
                            "errors": [{
                                "title": "Cannot serialize index metadata to JSON",
                                "detail": e.to_string()
                            }]
                        });
                        json_error.to_string()
                    },
                };

                match metadata {
                    Some(m) => {
                        let created_at = util::format_systemtime(m.created_at, "%+");
                        eprintln!("Created at: {created_at}");
                        if updater.has_updates(&m).await.map_err(|e| {
                            CliError::Network(format!("Geonames update check failed: {e}"))
                        })? {
                            winfo!(
                                "Updates available at Geonames.org. Use `qsv geocode \
                                 index-update` to update/rebuild the index.\nPlease use this \
                                 judiciously as Geonames is a free service.\n"
                            );
                        } else {
                            winfo!("Geonames index up-to-date.\n");
                        }
                    },
                    None => return fail_incorrectusage_clierror!("Invalid Geonames index file."),
                }

                // print to stdout the index metadata as JSON
                // so users can redirect stdout to a JSON file if desired
                println!("{index_metadata_json}");
            },
            GeocodeSubCmd::IndexUpdate => {
                // update/rebuild Geonames index from Geonames website
                // will only update if there are changes unless --force is specified
                check_index_file(&geocode_index_file)?;

                if args.flag_force {
                    display_rebuild_instructions(
                        &args.flag_cities_url,
                        &cities_filename,
                        &args.flag_languages,
                        &geocode_index_file,
                    );
                } else {
                    winfo!("Checking main Geonames website for updates...");
                    let metadata = index_storage
                        .read_metadata(geocode_index_file.clone())
                        .map_err(|e| format!("index-update error: {e}"))?;

                    let Some(m) = metadata else {
                        return fail_incorrectusage_clierror!("Invalid Geonames index file.");
                    };
                    if updater.has_updates(&m).await.map_err(|e| {
                        CliError::Network(format!("Geonames update check failed: {e}"))
                    })? {
                        winfo!("Updates available at Geonames.org.");
                        display_rebuild_instructions(
                            &args.flag_cities_url,
                            &cities_filename,
                            &args.flag_languages,
                            &geocode_index_file,
                        );
                    } else {
                        winfo!("Skipping update. Geonames index is up-to-date.");
                    }
                }
            },
            GeocodeSubCmd::IndexLoad => {
                // load alternate geocode index file
                if let Some(index_file) = args.arg_index_file {
                    winfo!("Validating alternate Geonames index: {index_file}...");
                    check_index_file(&index_file)?;

                    let engine_data =
                        load_engine_data(index_file.clone().into(), &progress).await?;
                    // we successfully loaded the alternate geocode index file, so its valid
                    // copy it to the default geocode index file

                    if engine_data.metadata.is_some() {
                        let _ =
                            index_storage.dump_to(active_geocode_index_file.clone(), &engine_data);
                        winfo!(
                            "Valid Geonames index file {index_file} successfully copied to \
                             {active_geocode_index_file}. It will be used from now on or until \
                             you reset/rebuild it.",
                        );
                    } else {
                        return fail_incorrectusage_clierror!(
                            "Alternate Geonames index file {index_file} is invalid.",
                        );
                    }
                } else {
                    return fail_incorrectusage_clierror!(
                        "No alternate Geonames index file specified."
                    );
                }
            },
            GeocodeSubCmd::IndexReset => {
                // reset geocode index by deleting the current local copy
                // and downloading the default geocode index for the current qsv version
                winfo!("Resetting Geonames index to default: {geocode_index_file}...");
                fs::remove_file(&geocode_index_file)?;
                load_engine_data(geocode_index_file.clone().into(), &progress).await?;
                winfo!("Default Geonames index file successfully reset to {QSV_VERSION} release.");
            },
            // index_cmd is true, so we should never get a non-index subcommand
            _ => unreachable!(),
        }
        return Ok(());
    }

    // we're not doing an index subcommand, so we're doing a suggest/now, reverse/now,
    // countryinfo/now or iplookup/now subcommand. Load the current local Geonames index
    let mut engine_data = load_engine_data(geocode_index_file.clone().into(), &progress).await?;
    if iplookup_cmd {
        // load the GeoIP2 database
        engine_data
            .load_geoip2(geoip2_filename.clone())
            .map_err(|e| {
                CliError::Other(format!(
                    r#"Error loading GeoIP2 database "{geoip2_filename}": {e}"#
                ))
            })?;
    }

    let engine = engine_data
        .as_engine()
        .map_err(|e| CliError::Other(format!("Error initializing Engine: {e}")))?;

    let mut rdr = rconfig.reader()?;
    let mut wtr = Config::new(args.flag_output.as_ref())
        .quote_style(
            // if we're doing a now subcommand with JSON output, we don't want the CSV writer
            // to close quote the output as it will produce invalid JSON
            if now_cmd && (args.flag_formatstr == "%json" || args.flag_formatstr == "%pretty-json")
            {
                csv::QuoteStyle::Never
            } else {
                csv::QuoteStyle::Necessary
            },
        )
        .writer()?;

    let headers = rdr.byte_headers()?.clone();
    let sel = rconfig.selection(&headers)?;
    let column_index = *sel.iter().next().unwrap();

    let mut headers = rdr.headers()?.clone();

    if let Some(new_name) = args.flag_rename {
        let new_col_names = util::ColumnNameParser::new(&new_name).parse()?;
        if new_col_names.len() != sel.len() {
            return fail_incorrectusage_clierror!(
                "Number of new columns does not match input column selection."
            );
        }
        for (i, col_index) in sel.iter().enumerate() {
            headers = replace_column_value(&headers, *col_index, &new_col_names[i]);
        }
    }

    // setup output headers
    if let Some(new_column) = &args.flag_new_column {
        headers.push_field(new_column);
    }

    // if formatstr starts with "%dyncols:"", then we're using dynfmt to add columns.
    // To add columns, we enclose in curly braces a key:value pair for each column with
    // the key being the desired column name and the value being the CityRecord field
    // we want to add to the CSV
    // e.g. "%dyncols: {city_col:name}, {state_col:admin1}, {country_col:country}"
    // will add three columns to the CSV named city_col, state_col and country_col.

    // first, parse the formatstr to get the column names and values in parallel vectors
    let mut column_names = Vec::new();
    let mut column_values = Vec::new();
    // dyncols_len is the number of columns we're adding in dyncols mode
    // it also doubles as a flag to indicate if we're using dyncols mode
    // i.e. if dyncols_len > 0, we're using dyncols mode; 0 we're not
    let dyncols_len = if args.flag_formatstr.starts_with("%dyncols:") {
        for column in args.flag_formatstr[9..].split(',') {
            let column = column.trim();
            let column_key_value: Vec<&str> = column.split(':').collect();
            if column_key_value.len() == 2 {
                column_names.push(column_key_value[0].trim_matches('{'));
                column_values.push(column_key_value[1].trim_matches('}'));
            }
        }

        // now, validate the column values
        // the valid column values are in SORTED_VALID_DYNCOLS
        for column_value in &column_values {
            if SORTED_VALID_DYNCOLS.binary_search(column_value).is_err() {
                return fail_incorrectusage_clierror!(
                    "Invalid column value: {column_value}. Valid values are: \
                     {SORTED_VALID_DYNCOLS:?}"
                );
            }
        }

        // its valid, add the columns to the CSV headers
        for column in column_names {
            headers.push_field(column);
        }
        // dyncols_len is u8 (0..=255); reject excessive column counts to avoid silent truncation
        let Ok(len) = u8::try_from(column_values.len()) else {
            return fail_incorrectusage_clierror!(
                "Too many %dyncols columns: {} (max 255).",
                column_values.len()
            );
        };
        len
    } else {
        0_u8
    };

    // now, write the headers to the output CSV, unless its a now subcommand with JSON output
    if !(now_cmd && (args.flag_formatstr == "%json" || args.flag_formatstr == "%pretty-json")) {
        wtr.write_record(&headers)?;
    }

    // setup admin1 filter for Suggest/Now
    let mut admin1_code_prefix = String::new();
    let mut admin1_same_prefix = true;
    let mut flag_country = args.flag_country.clone();
    let admin1_filter_list = match geocode_cmd {
        GeocodeSubCmd::Suggest | GeocodeSubCmd::SuggestNow => {
            // admin1 filter: if all uppercase, search for admin1 code, else, search for admin1 name
            // see https://download.geonames.org/export/dump/admin1CodesASCII.txt for valid codes
            if let Some(admin1_list) = args.flag_admin1.clone() {
                // this regex matches admin1 codes (e.g. US.NY, JP.40, CN.23, HK.NYL, GG.6417214)
                let admin1_code_re = ADMIN1_CODE_REGEX();
                let admin1_list_work = Some(
                    admin1_list
                        .split(',')
                        .map(|s| {
                            let temp_s = s.trim();
                            let is_code_flag = admin1_code_re.is_match(temp_s);
                            Admin1Filter {
                                admin1_string: if is_code_flag {
                                    if admin1_same_prefix {
                                        // check if all admin1 codes have the same prefix
                                        if admin1_code_prefix.is_empty() {
                                            // first admin1 code, so set the prefix
                                            admin1_code_prefix = temp_s[0..3].to_string();
                                        } else if admin1_code_prefix != temp_s[0..3] {
                                            // admin1 codes have different prefixes, so we can't
                                            // infer the country from the admin1 code
                                            admin1_same_prefix = false;
                                        }
                                    }
                                    temp_s.to_string()
                                } else {
                                    // its an admin1 name, lowercase it
                                    // so we can do case-insensitive starts_with() comparisons
                                    temp_s.to_lowercase()
                                },
                                is_code:       is_code_flag,
                            }
                        })
                        .collect::<Vec<Admin1Filter>>(),
                );

                // if admin1 is set, country must also be set
                // however, if all admin1 codes have the same prefix, we can infer the country from
                // the admin1 codes. Otherwise, we can't infer the country from the
                // admin1 code, so we error out.
                if args.flag_admin1.is_some() && flag_country.is_none() {
                    if !admin1_code_prefix.is_empty() && admin1_same_prefix {
                        admin1_code_prefix.pop(); // remove the dot
                        flag_country = Some(admin1_code_prefix);
                    } else {
                        return fail_incorrectusage_clierror!(
                            "If --admin1 is set, --country must also be set unless admin1 codes \
                             are used with a common country prefix (e.g. US.CA,US.NY,US.OH, etc)."
                        );
                    }
                }
                admin1_list_work
            } else {
                None
            }
        },
        _ => {
            // reverse/now and countryinfo/now subcommands don't support admin1 filter
            if args.flag_admin1.is_some() {
                return fail_incorrectusage_clierror!(
                    "reverse/reversenow & countryinfo subcommands do not support the --admin1 \
                     filter option."
                );
            }
            None
        },
    }; // end setup admin1 filters

    // setup country filter - both suggest/now and reverse/now support country filters
    // countryinfo/now subcommands ignores the country filter
    let country_filter_list = flag_country.map(|country_list| {
        country_list
            .split(',')
            .map(|s| s.trim().to_ascii_uppercase())
            .collect::<Vec<String>>()
    });

    log::debug!("country_filter_list: {country_filter_list:?}");
    log::debug!("admin1_filter_list: {admin1_filter_list:?}");

    // amortize memory allocation by reusing record
    let mut batch_record = csv::StringRecord::new();

    // reuse batch buffers
    let batchsize: usize = if args.flag_batch == 0 {
        std::cmp::max(1000, util::count_rows_regular(&rconfig)? as usize)
    } else {
        args.flag_batch
    };
    let mut batch = Vec::with_capacity(batchsize);
    let mut batch_results = Vec::with_capacity(batchsize);

    util::njobs(args.flag_jobs);

    let invalid_result = args.flag_invalid_result.unwrap_or_default();

    let min_score = args.flag_min_score;
    let k_weight = args.flag_k_weight;

    // main loop to read CSV and construct batches for parallel processing.
    // each batch is processed via Rayon parallel iterator.
    // loop exits when batch is empty.
    'batch_loop: loop {
        for _ in 0..batchsize {
            match rdr.read_record(&mut batch_record) {
                Ok(has_data) => {
                    if has_data {
                        batch.push(std::mem::take(&mut batch_record));
                    } else {
                        // nothing else to add to batch
                        break;
                    }
                },
                Err(e) => {
                    return fail_clierror!("Error reading file: {e}");
                },
            }
        }

        if batch.is_empty() {
            // break out of infinite loop when at EOF
            break 'batch_loop;
        }

        // do actual apply command via Rayon parallel iterator
        batch
            .par_iter()
            .map(|record_item| {
                let mut record = record_item.clone();
                let mut cell = record[column_index].to_owned();
                if cell.is_empty() {
                    // cell to geocode is empty. If in dyncols mode, we need to add empty columns.
                    // Otherwise, we leave the row untouched.
                    if dyncols_len > 0 {
                        add_fields(&mut record, "", dyncols_len);
                    }
                } else if geocode_cmd == GeocodeSubCmd::CountryInfo
                    || geocode_cmd == GeocodeSubCmd::CountryInfoNow
                {
                    // we're doing a countryinfo or countryinfonow subcommand
                    cell = get_countryinfo(
                        &engine,
                        &cell.to_ascii_uppercase(),
                        &args.flag_language,
                        &args.flag_formatstr,
                    )
                    .unwrap_or(cell);
                } else if dyncols_len > 0 {
                    // we're in dyncols mode, so use search_index_NO_CACHE fn
                    // as we need to inject the column values into each row of the output csv
                    // so we can't use the cache
                    let search_results = search_index_uncached(
                        &engine,
                        geocode_cmd,
                        &cell,
                        &args.flag_formatstr,
                        &args.flag_language,
                        min_score,
                        k_weight,
                        country_filter_list.as_deref(),
                        admin1_filter_list.as_deref(),
                        &column_values,
                        &mut record,
                    );

                    // if search_results.is_some but we don't get the DYNCOLS_POPULATED
                    // sentinel value or its None, then we have an invalid result
                    let invalid = if let Some(res) = search_results {
                        res != DYNCOLS_POPULATED
                    } else {
                        true
                    };
                    if invalid {
                        if invalid_result.is_empty() {
                            // --invalid-result is not set, so add empty columns
                            add_fields(&mut record, "", dyncols_len);
                        } else {
                            // --invalid-result is set
                            add_fields(&mut record, &invalid_result, dyncols_len);
                        }
                    }
                } else {
                    // not in dyncols mode so call the CACHED search_index fn
                    // as we want to take advantage of the cache
                    let search_result = search_index(
                        &engine,
                        geocode_cmd,
                        &cell,
                        &args.flag_formatstr,
                        &args.flag_language,
                        min_score,
                        k_weight,
                        country_filter_list.as_deref(),
                        admin1_filter_list.as_deref(),
                        &column_values,
                        &mut record,
                    );

                    if let Some(geocoded_result) = search_result {
                        // we have a valid geocode result, so use that
                        cell = geocoded_result;
                    } else {
                        // we have an invalid geocode result
                        if !invalid_result.is_empty() {
                            // --invalid-result is set, so use that instead
                            // otherwise, we leave cell untouched.
                            cell.clone_from(&invalid_result);
                        }
                    }
                }
                if args.flag_new_column.is_some() {
                    record.push_field(&cell);
                } else {
                    record = replace_column_value(&record, column_index, &cell);
                }

                record
            })
            .collect_into_vec(&mut batch_results);

        // rayon collect() guarantees original order, so we can just append results each batch
        for result_record in &batch_results {
            wtr.write_record(result_record)?;
        }

        if show_progress {
            progress.inc(batch.len() as u64);
        }

        batch.clear();
    } // end batch loop

    if show_progress {
        // the geocode result cache is NOT used in dyncols mode,
        // so update the cache info only when dyncols_len == 0.
        // SEARCH_INDEX is a sharded ConcurrentCached store (not the RwLock-wrapped
        // #[cached] store that util::update_cache_info! reads via .read()), so pull its
        // stats from the store's own metrics() snapshot.
        if dyncols_len == 0 {
            let metrics = SEARCH_INDEX.metrics();
            let entry_count = metrics.entry_count.unwrap_or(0);
            if entry_count > 0 {
                #[allow(clippy::cast_precision_loss)]
                progress.set_message(format!(
                    " of {} records. Cache {:.2}% entries: {} capacity: {}.",
                    indicatif::HumanCount(progress.length().unwrap()),
                    metrics.hit_ratio().unwrap_or(0.0) * 100.0,
                    indicatif::HumanCount(entry_count as u64),
                    indicatif::HumanCount(metrics.capacity.unwrap_or(0) as u64),
                ));
            }
        }
        util::finish_progress(&progress);
    }
    Ok(wtr.flush()?)
}

/// Run the `opencage` / `opencagenow` subcommands: online forward/reverse geocoding
/// via the `OpenCage` API. Rows are processed sequentially behind a rate limiter, with
/// a persistent on-disk result cache (`OpenCage`'s TOS explicitly permits caching).
/// This is a separate path from `geocode_main`'s offline, rayon-parallel pipeline.
#[allow(clippy::future_not_send)]
async fn run_opencage(args: Args, mode: GeocodeSubCmd, cache_dir: &Path) -> CliResult<()> {
    let now_cmd = mode == GeocodeSubCmd::OpencageNow;

    // for opencagenow, write the single CLI value to a one-row temp CSV, so the rest
    // of the pipeline (reader, headers, JSON quoting) is identical to opencage
    let tempdir = tempfile::Builder::new().prefix("qsv-geocode").tempdir()?;
    let input = if now_cmd {
        let temp_csv_path = format!(
            "{}/{}.csv",
            tempdir.path().to_string_lossy(),
            Uuid::new_v4()
        );
        let mut temp_csv_wtr = csv::WriterBuilder::new().from_path(&temp_csv_path)?;
        temp_csv_wtr.write_record(["Location"])?;
        temp_csv_wtr.write_record([&args.arg_location])?;
        temp_csv_wtr.flush()?;
        Some(temp_csv_path)
    } else {
        args.arg_input.clone()
    };

    let rconfig = Config::new(input.as_ref())
        .delimiter(args.flag_delimiter)
        .select(SelectColumns::parse(&args.arg_column)?);

    // parse & validate the "%dyncols:" formatstr (if used). In dyncols mode, one
    // output column is added per "{col_name:key}" pair, where key is an OpenCage
    // dynamic field key (formatted, lat, lng, confidence, components.* or
    // annotations.*). Unlike the predefined & dynamic formats, dyncols mode does
    // not replace the input column.
    let dyncols_mode = args.flag_formatstr.starts_with("%dyncols:");
    let mut column_names: Vec<String> = Vec::new();
    let mut column_values: Vec<String> = Vec::new();
    if dyncols_mode {
        for column in args.flag_formatstr[9..].split(',') {
            let column = column.trim();
            if column.is_empty() {
                // tolerate a trailing/empty comma-delimited entry
                continue;
            }
            let column_key_value: Vec<&str> = column.split(':').collect();
            if column_key_value.len() != 2 {
                return fail_incorrectusage_clierror!(
                    "Invalid '%dyncols:' pair: {column:?}. Expected a single '{{col_name:key}}' \
                     pair."
                );
            }
            let column_name = column_key_value[0].trim_matches('{').trim();
            let column_value = column_key_value[1].trim_matches('}').trim();
            if column_name.is_empty() {
                return fail_incorrectusage_clierror!(
                    "Invalid '%dyncols:' pair: {column:?}. The column name is empty."
                );
            }
            column_names.push(column_name.to_string());
            column_values.push(column_value.to_string());
        }
        if column_values.is_empty() {
            return fail_incorrectusage_clierror!(
                "Invalid '%dyncols:' format - expected one or more 'col_name:key' pairs enclosed \
                 in curly braces."
            );
        }
        for column_value in &column_values {
            if !is_valid_opencage_dyncol(column_value) {
                return fail_incorrectusage_clierror!(
                    "Invalid '%dyncols:' key: {column_value}. Valid keys are: formatted, lat, \
                     lng, confidence, components.<name> or annotations.<dotted.path>."
                );
            }
        }
    }
    // dyncols_len doubles as the empty/invalid fill count; u8 mirrors geocode_main
    let Ok(dyncols_len) = u8::try_from(column_values.len()) else {
        return fail_incorrectusage_clierror!(
            "Too many %dyncols columns: {} (max 255).",
            column_values.len()
        );
    };

    #[cfg(feature = "datapusher_plus")]
    let show_progress = false;
    #[cfg(not(feature = "datapusher_plus"))]
    let show_progress =
        (args.flag_progressbar || util::get_envvar_flag("QSV_PROGRESSBAR")) && !rconfig.is_stdin();

    let progress = ProgressBar::with_draw_target(None, ProgressDrawTarget::stderr_with_hz(5));
    if show_progress {
        util::prep_progress(&progress, util::count_rows(&rconfig)?);
    } else {
        progress.set_draw_target(ProgressDrawTarget::hidden());
    }

    // when a now subcommand outputs JSON, the CSV writer must not quote the output
    // (and the "Location" header is omitted) so the result is valid JSON
    let json_output =
        now_cmd && (args.flag_formatstr == "%json" || args.flag_formatstr == "%pretty-json");

    let mut rdr = rconfig.reader()?;
    let mut wtr = Config::new(args.flag_output.as_ref())
        .quote_style(if json_output {
            csv::QuoteStyle::Never
        } else {
            csv::QuoteStyle::Necessary
        })
        .writer()?;

    let header_record = rdr.byte_headers()?.clone();
    let sel = rconfig.selection(&header_record)?;
    let column_index = *sel.iter().next().unwrap();

    let mut headers = rdr.headers()?.clone();
    if let Some(new_name) = &args.flag_rename {
        let new_col_names = util::ColumnNameParser::new(new_name).parse()?;
        if new_col_names.len() != sel.len() {
            return fail_incorrectusage_clierror!(
                "Number of new columns does not match input column selection."
            );
        }
        for (i, col_index) in sel.iter().enumerate() {
            headers = replace_column_value(&headers, *col_index, &new_col_names[i]);
        }
    }
    if let Some(new_column) = &args.flag_new_column {
        headers.push_field(new_column);
    }
    // dyncols mode: append one header per "{col_name:key}" pair
    for column_name in &column_names {
        headers.push_field(column_name);
    }
    if !json_output {
        wtr.write_record(&headers)?;
    }

    // build the async HTTP client (rustls, honors --timeout)
    let timeout = std::time::Duration::from_secs(util::timeout_secs(args.flag_timeout)?);
    let client = reqwest::Client::builder()
        .user_agent(util::set_user_agent(None)?)
        .timeout(timeout)
        .use_rustls_tls()
        .brotli(true)
        .gzip(true)
        .deflate(true)
        .zstd(true)
        .build()?;

    // rate limiter - allow_burst(1) avoids an initial burst (see governor#39)
    // safety: flag_rate_limit is validated to be >= 1 in run()
    let rate_limit = NonZeroU32::new(args.flag_rate_limit).unwrap();
    let limiter =
        RateLimiter::direct(Quota::per_second(rate_limit).allow_burst(NonZeroU32::new(1).unwrap()));

    // persistent on-disk result cache (OpenCage's TOS explicitly permits caching)
    let disk_cache = if args.flag_no_cache {
        None
    } else {
        let mut cache_builder = RedbCache::builder("geocode-opencage")
            .disk_dir(cache_dir)
            .refresh_on_hit(false)
            // preserve v2/sled non-durable behavior: OpenCage responses are
            // recomputable external-API results, and fsync-per-write on the
            // per-row geocoding hot path is the throughput hit we avoid.
            .durable(false);
        // --cache-ttl 0 disables time-based expiration (entries cached
        // indefinitely). v3's RedbCache builder rejects .ttl(0), so only set a
        // TTL when it is non-zero; leaving it unset means "never expire".
        if args.flag_cache_ttl != 0 {
            cache_builder = cache_builder.ttl(std::time::Duration::from_secs(args.flag_cache_ttl));
        }
        let cache = cache_builder
            .build()
            .map_err(|e| CliError::Other(format!("Error building OpenCage disk cache: {e}")))?;
        if let Err(e) = cache.remove_expired_entries() {
            log::warn!("Could not remove expired OpenCage cache entries: {e}");
        }
        Some(cache)
    };
    let mut mem_cache: HashMap<String, String> = HashMap::new();

    // safety: flag_api_key is resolved (flag > env) and validated non-empty in run()
    let api_key = args.flag_api_key.as_deref().unwrap_or_default();
    let invalid_result = args.flag_invalid_result.clone().unwrap_or_default();
    let formatstr = args.flag_formatstr.as_str();
    let language = if args.flag_language.is_empty() {
        "en"
    } else {
        args.flag_language.as_str()
    };
    let countrycode = args.flag_country.as_deref();

    let mut record = csv::StringRecord::new();
    while rdr.read_record(&mut record)? {
        let cell = record.get(column_index).unwrap_or_default().to_string();

        if dyncols_mode {
            // dyncols mode: keep the input row intact and append one field per
            // requested column, geocoding (with caching) when the cell is non-empty
            let mut out_record = record.clone();
            let values = if cell.trim().is_empty() {
                // empty cell - no geocoding
                None
            } else if let Some(query) = normalize_opencage_query(&cell, args.flag_reverse) {
                match opencage_lookup_dyncols(
                    &client,
                    &limiter,
                    disk_cache.as_ref(),
                    &mut mem_cache,
                    api_key,
                    &query,
                    language,
                    countrycode,
                    args.flag_no_annotations,
                    &column_values,
                )
                .await
                {
                    Ok(values) => values,
                    Err(OcError::Fatal(msg)) => return fail_clierror!("{msg}"),
                    Err(OcError::Transient(msg)) => {
                        log::warn!("OpenCage lookup failed for {cell:?}: {msg}");
                        None
                    },
                }
            } else {
                // --reverse is set but the cell is not a WGS-84 coordinate
                None
            };
            if let Some(values) = values {
                for value in &values {
                    out_record.push_field(value);
                }
            } else {
                // empty cell, non-coordinate or no result: fill the added columns
                // with --invalid-result (an empty string if --invalid-result is unset)
                add_fields(&mut out_record, &invalid_result, dyncols_len);
            }
            wtr.write_record(&out_record)?;
        } else {
            let out_cell = if cell.trim().is_empty() {
                // empty cell - leave the row untouched
                cell.clone()
            } else if let Some(query) = normalize_opencage_query(&cell, args.flag_reverse) {
                match opencage_lookup(
                    &client,
                    &limiter,
                    disk_cache.as_ref(),
                    &mut mem_cache,
                    api_key,
                    &query,
                    formatstr,
                    language,
                    countrycode,
                    args.flag_no_annotations,
                )
                .await
                {
                    Ok(Some(result)) => result,
                    Ok(None) if invalid_result.is_empty() => cell.clone(),
                    Ok(None) => invalid_result.clone(),
                    Err(OcError::Fatal(msg)) => return fail_clierror!("{msg}"),
                    Err(OcError::Transient(msg)) => {
                        log::warn!("OpenCage lookup failed for {cell:?}: {msg}");
                        if invalid_result.is_empty() {
                            cell.clone()
                        } else {
                            invalid_result.clone()
                        }
                    },
                }
            } else if invalid_result.is_empty() {
                // --reverse is set but the cell is not a WGS-84 coordinate
                cell.clone()
            } else {
                invalid_result.clone()
            };

            let out_record = if args.flag_new_column.is_some() {
                let mut new_record = record.clone();
                new_record.push_field(&out_cell);
                new_record
            } else {
                replace_column_value(&record, column_index, &out_cell)
            };
            wtr.write_record(&out_record)?;
        }
        if show_progress {
            progress.inc(1);
        }
    }

    if show_progress {
        util::finish_progress(&progress);
    }
    Ok(wtr.flush()?)
}

// ─────────────────────── OpenCage disk-cache management ───────────────────────

/// The cache name used for the persistent on-disk `OpenCage` result cache.
/// `cached`'s `RedbCache` stores the redb database file under `{cache_dir}`,
/// naming it after this cache name.
const OPENCAGE_CACHE_NAME: &str = "geocode-opencage";

/// Parse a relative age like "30d", "2w", "48h", "90m", "3600s" (case-insensitive,
/// surrounding whitespace allowed) into a Duration. Returns None when the input is
/// not a relative-age string, so the caller can fall back to absolute-date parsing.
fn parse_relative_age(input: &str) -> Option<Duration> {
    let caps = RELATIVE_AGE_REGEX().captures(input.trim())?;
    let n: u64 = caps.get(1)?.as_str().parse().ok()?;
    let secs = match caps.get(2)?.as_str().to_ascii_lowercase().as_str() {
        "s" => n,
        "m" => n.checked_mul(60)?,
        "h" => n.checked_mul(3_600)?,
        "d" => n.checked_mul(86_400)?,
        "w" => n.checked_mul(604_800)?,
        _ => return None,
    };
    Some(Duration::from_secs(secs))
}

/// Resolve a `--older-than` value into an age threshold (the lifespan handed to
/// `RedbCacheBuilder::ttl`). Accepts a relative age (30d, 2w, ...) or an
/// absolute date/datetime parsed by `qsv_dateparser`. A future cutoff is rejected.
fn resolve_older_than(value: &str) -> CliResult<Duration> {
    if let Some(age) = parse_relative_age(value) {
        return Ok(age);
    }
    // a value with the shape of a relative age that didn't parse overflowed -
    // report that explicitly instead of misleadingly falling through to date parsing
    if RELATIVE_AGE_REGEX().is_match(value.trim()) {
        return Err(CliError::IncorrectUsage(format!(
            "Invalid --older-than value '{value}': the relative age is too large."
        )));
    }
    // not a relative age - try parsing as an absolute date/datetime
    let cutoff = qsv_dateparser::parse(value).map_err(|e| {
        CliError::IncorrectUsage(format!(
            "Invalid --older-than value '{value}': not a relative age (e.g. 30d, 2w) nor a \
             parseable date - {e}"
        ))
    })?;
    let cutoff_secs = u64::try_from(cutoff.timestamp()).map_err(|_| {
        CliError::IncorrectUsage(format!(
            "Invalid --older-than value '{value}': date is before the Unix epoch."
        ))
    })?;
    let cutoff_systime = SystemTime::UNIX_EPOCH + Duration::from_secs(cutoff_secs);
    SystemTime::now()
        .duration_since(cutoff_systime)
        .map_err(|_| {
            CliError::IncorrectUsage(format!(
                "Invalid --older-than value '{value}': the date is in the future."
            ))
        })
}

/// Run the `cache-clear` / `cache-prune` / `cache-info` subcommands, which manage
/// the persistent on-disk `OpenCage` result cache. Synchronous - does not use the
/// Geonames index nor make any network calls.
fn run_cache_mgmt(args: &Args, mode: GeocodeSubCmd, cache_dir: &Path) -> CliResult<()> {
    // graceful "no cache" path - detect an existing cache WITHOUT building one
    // (building a RedbCache would create an empty redb file as a side effect).
    // cached names the redb file after the cache name, so scan cache_dir for it.
    let cache_exists = std::fs::read_dir(cache_dir).ok().is_some_and(|entries| {
        entries.flatten().any(|e| {
            e.file_name()
                .to_string_lossy()
                .starts_with(OPENCAGE_CACHE_NAME)
        })
    });

    if !cache_exists {
        winfo!(
            "No OpenCage cache found in {}. Nothing to do.",
            cache_dir.display()
        );
        if mode == GeocodeSubCmd::CacheInfo {
            println!(
                "{}",
                json!({
                    "cache_dir":   cache_dir.display().to_string(),
                    "exists":      false,
                    "entry_count": 0,
                })
            );
        }
        return Ok(());
    }

    // TTL is only meaningful for cache-prune (it is the age threshold used by
    // remove_expired_entries). v3's RedbCache builder rejects a zero TTL, and TTL
    // is optional, so only set it for prune; clear/info open the cache without one.
    let mut builder = RedbCache::builder(OPENCAGE_CACHE_NAME)
        .disk_dir(cache_dir)
        .refresh_on_hit(false);
    if mode == GeocodeSubCmd::CachePrune {
        // run() guarantees --older-than is present & valid for cache-prune
        let lifespan = resolve_older_than(args.flag_older_than.as_deref().unwrap_or_default())?;
        // a zero age threshold would mean "prune every entry" - that overlaps
        // cache-clear, and v3's RedbCache builder rejects a zero TTL anyway, so
        // reject it here with a pointer to the right subcommand.
        if lifespan.is_zero() {
            return Err(CliError::IncorrectUsage(
                "--older-than resolves to a zero age threshold. To remove every entry, use `qsv \
                 geocode cache-clear` instead."
                    .to_string(),
            ));
        }
        // fsync the removals durably to disk
        builder = builder.ttl(lifespan).durable(true);
    }
    let cache: RedbCache<String, String> = builder
        .build()
        .map_err(|e| CliError::Other(format!("Error opening OpenCage disk cache: {e}")))?;

    match mode {
        GeocodeSubCmd::CacheClear => {
            // RedbCache cannot report a live entry count cheaply (cache_size() is
            // Ok(None)), so we clear without claiming a (misleading) removed count.
            cache
                .cache_clear()
                .map_err(|e| CliError::Other(format!("Error clearing OpenCage cache: {e}")))?;
            cache
                .flush()
                .map_err(|e| CliError::Other(format!("Error flushing OpenCage cache: {e}")))?;
            winfo!("Cleared the OpenCage cache.");
        },
        GeocodeSubCmd::CachePrune => {
            // remove_expired_entries() returns a real swept count (it scans), unlike
            // cache_size() which RedbCache reports as Ok(None).
            let pruned = cache
                .remove_expired_entries()
                .map_err(|e| CliError::Other(format!("Error pruning OpenCage cache: {e}")))?;
            cache
                .flush()
                .map_err(|e| CliError::Other(format!("Error flushing OpenCage cache: {e}")))?;
            winfo!("Pruned {pruned} expired entries from the OpenCage cache.");
        },
        GeocodeSubCmd::CacheInfo => {
            // RedbCache::cache_size() is Ok(None) (an exact count is an O(n) scan it
            // won't pay implicitly), so entry_count is reported as unavailable rather
            // than a misleading 0.
            let entry_count = cache
                .cache_size()
                .map_err(|e| CliError::Other(format!("Error reading OpenCage cache: {e}")))?;
            let disk_path = cache.disk_path();
            // metadata() can fail on I/O errors; degrade gracefully rather than
            // reporting a misleading 0 bytes (None -> "unknown" / JSON null)
            let size_on_disk: Option<u64> = std::fs::metadata(disk_path).ok().map(|m| m.len());
            let size_human = size_on_disk.map(|s| HumanBytes(s).to_string());

            winfo!("OpenCage cache file: {}", disk_path.display());
            winfo!(
                "Entries: {}",
                entry_count.map_or_else(|| "unknown".to_string(), |n| n.to_string())
            );
            winfo!(
                "Size on disk: {}",
                size_human.clone().unwrap_or_else(|| "unknown".to_string())
            );

            println!(
                "{}",
                json!({
                    "cache_dir":    cache_dir.display().to_string(),
                    "cache_file":   disk_path.display().to_string(),
                    "exists":       true,
                    "entry_count":  entry_count,
                    "size_on_disk": size_on_disk,
                    "size_human":   size_human,
                })
            );
        },
        _ => unreachable!("run_cache_mgmt called with a non-cache subcommand"),
    }
    Ok(())
}

/// Display instructions for rebuilding the Geonames index using the geosuggest crate directly
fn display_rebuild_instructions(
    cities_url: &str,
    cities_filename: &str,
    languages: &str,
    geocode_index_file: &str,
) {
    winfo!(
        r"To rebuild the index, use the geosuggest crate directly:

git clone https://github.com/estin/geosuggest.git
cd geosuggest
cargo run -p geosuggest-utils --bin geosuggest-build-index --release --features=cli,tracing -- \
    from-urls \
    --cities-url {cities_url} \
    --cities-filename {cities_filename} \
    --languages {languages} \
    --output {geocode_index_file}",
        cities_url = cities_url,
        cities_filename = cities_filename,
        languages = languages,
        geocode_index_file = geocode_index_file,
    );
}

/// check if `index_file` exists and ends with a .rkyv extension
fn check_index_file(index_file: &str) -> CliResult<()> {
    // the only prebuilt index published to the qsv GitHub repo is cities15000,
    // so 15000 is the sole numeric shortcut accepted by the index-load subcommand
    if let Ok(i) = index_file.parse::<u16>() {
        if i == 15000 {
            return Ok(());
        }
        return fail_incorrectusage_clierror!(
            "Invalid index shortcut {index_file} - only 15000 is supported (the prebuilt \
             cities15000 index). To use a different city set, rebuild the index with \
             `index-update --cities-url`."
        );
    }

    if !std::path::Path::new(index_file)
        .extension()
        .is_some_and(|ext| ext.eq_ignore_ascii_case("rkyv"))
    {
        return fail_incorrectusage_clierror!(
            "Alternate Geonames index file {index_file} does not have a .rkyv extension."
        );
    }
    // check if index_file exist
    if !Path::new(index_file).exists() {
        return fail_incorrectusage_clierror!(
            "Alternate Geonames index file {index_file} does not exist."
        );
    }

    winfo!("Valid: {index_file}");
    Ok(())
}

/// `load_engine_data` loads the Geonames index file into memory
/// if the index file does not exist, it will download the default index file
/// from the qsv GitHub repo. For convenience, if `geocode_index_file` is the bare
/// number 15000 (no extension), it downloads the prebuilt cities15000 index instead.
async fn load_engine_data(
    geocode_index_file: PathBuf,
    progressbar: &ProgressBar,
) -> CliResult<EngineData> {
    // default cities index file
    static DEFAULT_GEONAMES_CITIES_INDEX: u16 = 15000;

    let index_file = std::path::Path::new(&geocode_index_file);

    // the file stem is used to detect the numeric shortcut (e.g. `index-load 15000`)
    let geocode_index_file_stem = geocode_index_file
        .file_stem()
        .map(|s| s.to_string_lossy().to_string())
        .unwrap_or_default();

    let download_url = format!(
        "https://github.com/dathere/qsv/releases/download/{QSV_VERSION}/{DEFAULT_GEOCODE_INDEX_FILENAME}.cities"
    );

    // the numeric shortcut (e.g. `index-load 15000`) only applies to a bare number
    // with no file extension - a real local file like `15000.rkyv` must be loaded
    // as-is, not mistaken for the shortcut and overwritten by a download.
    let numeric_shortcut = if geocode_index_file.extension().is_none() {
        geocode_index_file_stem.parse::<u16>().ok()
    } else {
        None
    };

    if let Some(shortcut) = numeric_shortcut {
        // the only prebuilt index in the qsv GitHub repo is cities15000
        if shortcut != DEFAULT_GEONAMES_CITIES_INDEX {
            return fail_incorrectusage_clierror!(
                "Only the prebuilt cities15000 index is supported via the numeric shortcut (use \
                 15000). To use a different city set, rebuild the index with `index-update \
                 --cities-url`."
            );
        }

        progressbar.println(format!(
            "Downloading the prebuilt cities15000 Geonames index for qsv {QSV_VERSION} release..."
        ));

        util::download_file(
            &format!("{download_url}{shortcut}.sz"),
            geocode_index_file.clone(),
            !progressbar.is_hidden(),
            None,
            Some(60),
            None,
        )
        .await?;
    } else if index_file.exists() {
        // load existing local index
        progressbar.println(format!(
            "Loading existing Geonames index from {}",
            index_file.display()
        ));
    } else {
        // initial load or index-reset, download index file from qsv releases
        progressbar.println(format!(
            "Downloading default Geonames index for qsv {QSV_VERSION} release..."
        ));

        util::download_file(
            &format!("{download_url}{DEFAULT_GEONAMES_CITIES_INDEX}"),
            geocode_index_file.clone(),
            !progressbar.is_hidden(),
            None,
            Some(60),
            None,
        )
        .await?;
    }

    // check if the geocode_index_file is snappy compressed; decompress it if so.
    // detect by extension (case-insensitive) OR by snappy framed-format magic header,
    // since the numeric-shortcut download saves a .sz payload to a path without an
    // extension (e.g. "500"), and a user-supplied path may also lose the suffix.
    let has_snappy_extension = geocode_index_file
        .extension()
        .and_then(|ext| ext.to_str())
        .is_some_and(|ext| ext.eq_ignore_ascii_case("sz"));
    let is_snappy_framed_file = {
        let mut header = [0_u8; 10];
        fs::File::open(&geocode_index_file)
            .and_then(|mut file| std::io::Read::read_exact(&mut file, &mut header))
            .is_ok_and(|()| header == *b"\xff\x06\x00\x00sNaPpY")
    };
    let geocode_index_file = if has_snappy_extension || is_snappy_framed_file {
        // if the file is `foo.rkyv.sz`, drop only the `.sz` suffix to get `foo.rkyv`;
        // otherwise replace/add the extension with `.rkyv`. Note: `with_extension(".rkyv")`
        // would yield `foo..rkyv` because the leading dot is treated as part of the
        // raw extension argument.
        let decompressed_geocode_index_file =
            if geocode_index_file.file_stem().is_some_and(|stem| {
                std::path::Path::new(stem)
                    .extension()
                    .is_some_and(|ext| ext == "rkyv")
            }) {
                geocode_index_file.with_extension("")
            } else {
                geocode_index_file.with_extension("rkyv")
            };
        progressbar.println(format!(
            "Decompressing {} to {}",
            geocode_index_file.display(),
            decompressed_geocode_index_file.display()
        ));
        let tmpdir = tempdir()?;
        let decompressed_tmpfile = util::decompress_snappy_file(&geocode_index_file, &tmpdir)?;
        fs::copy(decompressed_tmpfile, &decompressed_geocode_index_file)?;
        decompressed_geocode_index_file
    } else {
        geocode_index_file
    };

    let storage = storage::Storage::new();

    let engine = storage
        .load_from(geocode_index_file)
        .map_err(|e| format!("On load index file: {e}"))?;

    if let Some(metadata) = &engine.metadata {
        let now = std::time::SystemTime::now();
        let age = now.duration_since(metadata.created_at).unwrap();
        let created_at_formatted = util::format_systemtime(metadata.created_at, "%+");

        progressbar.println(format!(
            "Geonames index loaded. Created: {created_at_formatted}  Age: {}",
            indicatif::HumanDuration(age)
        ));
    }

    Ok(engine)
}

/// `search_index` returns a geocode result for a given cell value, used by the
/// suggest/suggestnow and reverse/reversenow subcommands. It wraps
/// `search_index_uncached` with a cache keyed on the cell value, storing the
/// formatted geocoded result. We CANNOT use the cache in dyncols mode (the cached
/// result is the formatted string, not the individual fields), so dyncols mode calls
/// `search_index_uncached` directly.
///
/// `search_index` wraps `search_index_uncached` with a *sharded* concurrent LRU cache
/// (cached 2.0 `#[concurrent_cached]`) keyed on the cell value. This replaces the
/// single-`RwLock` `#[cached]` `LruCache`, whose per-hit recency-bump write lock
/// serialized geocode's rayon-parallel pipeline (on a 16-core box this is ~2.2x faster
/// on a high-hit-ratio workload). `ShardedLruCache` still takes a write lock per read
/// hit, but spreads it across many independent shards. With `shards` omitted, the store
/// derives the shard count from `std::thread::available_parallelism()` (`4 x cores`,
/// rounded to a power of two and clamped to [8, 1024]), so it auto-scales to the host.
/// `max_size` must be an integer literal (the macro parses it as a literal), so the
/// `2_000_000` cap (~2M entries) is inlined here. The cap is split across shards (each
/// shard's map allocates lazily), so the build can't fail on capacity — no
/// graceful-fallback path is needed.
#[concurrent_cached(
    max_size = 2_000_000,
    key = "String",
    convert = r#"{ cell.to_owned() }"#
)]
fn search_index(
    engine: &Engine,
    mode: GeocodeSubCmd,
    cell: &str,
    formatstr: &str,
    lang_lookup: &str,
    min_score: Option<f32>,
    k: Option<f32>,
    country_filter_list: Option<&[String]>,
    admin1_filter_list: Option<&[Admin1Filter]>,
    column_values: &[&str],
    record: &mut csv::StringRecord,
) -> Option<String> {
    search_index_uncached(
        engine,
        mode,
        cell,
        formatstr,
        lang_lookup,
        min_score,
        k,
        country_filter_list,
        admin1_filter_list,
        column_values,
        record,
    )
}

/// Uncached geocode lookup — the raw body shared by the cached `search_index`
/// wrapper and dyncols mode (which cannot use the cache).
fn search_index_uncached(
    engine: &Engine,
    mode: GeocodeSubCmd,
    cell: &str,
    formatstr: &str,
    lang_lookup: &str,
    min_score: Option<f32>,
    k: Option<f32>,
    country_filter_list: Option<&[String]>,
    admin1_filter_list: Option<&[Admin1Filter]>,
    column_values: &[&str],
    record: &mut csv::StringRecord,
) -> Option<String> {
    if mode == GeocodeSubCmd::Suggest || mode == GeocodeSubCmd::SuggestNow {
        let search_result: Vec<&CitiesRecord>;
        let cityrecord = if let Some(admin1_filter_list) = admin1_filter_list {
            // we have an admin1 filter, run a search for top SUGGEST_ADMIN1_LIMIT results
            search_result =
                engine.suggest(cell, SUGGEST_ADMIN1_LIMIT, min_score, country_filter_list);

            // first, get the first result and store that in cityrecord
            let Some(first_result) = search_result.first().copied() else {
                // no results, so return early with None
                return None;
            };

            // then iterate through search results and find the first one that matches admin1
            // the search results are already sorted by score, so we just need to find the first
            let mut matched_record: Option<&CitiesRecord> = None;
            'outer: for cr in &search_result {
                if let Some(admin_division) = cr.admin_division.as_ref() {
                    // lowercase the admin1 name once per candidate, only when needed
                    let mut admin1_name_lower: Option<String> = None;
                    for admin1_filter in admin1_filter_list {
                        if admin1_filter.is_code {
                            // admin1 is a code, so we search for admin1 code
                            if admin_division
                                .code
                                .starts_with(&admin1_filter.admin1_string)
                            {
                                matched_record = Some(cr);
                                break 'outer;
                            }
                        } else {
                            // admin1 is a name, case-insensitive starts_with
                            let lower = admin1_name_lower
                                .get_or_insert_with(|| admin_division.name.to_lowercase());
                            if lower.starts_with(&admin1_filter.admin1_string) {
                                matched_record = Some(cr);
                                break 'outer;
                            }
                        }
                    }
                }
            }

            // no admin1 match, so we return the first result
            matched_record.unwrap_or(first_result)
        } else {
            // no admin1 filter, run a search for 1 result (top match)
            search_result = engine.suggest(cell, 1, min_score, country_filter_list);
            let Some(cr) = search_result.into_iter().next() else {
                // no results, so return early with None
                return None;
            };
            cr
        };

        let country = cityrecord.country.as_ref()?.code.as_str();

        let nameslang = get_cityrecord_name_in_lang(cityrecord, lang_lookup);

        if formatstr == "%+" {
            // default for suggest is location - e.g. "(lat, long)"
            if mode == GeocodeSubCmd::SuggestNow {
                // however, make SuggestNow default more verbose
                return Some(format!(
                    "{name}, {admin1name} {country}: {latitude}, {longitude}",
                    name = nameslang.cityname,
                    admin1name = nameslang.admin1name,
                    latitude = cityrecord.latitude,
                    longitude = cityrecord.longitude
                ));
            }
            return Some(format!(
                "({latitude}, {longitude})",
                latitude = cityrecord.latitude,
                longitude = cityrecord.longitude
            ));
        }

        let capital = engine
            .capital(country)
            .map(|cr| cr.name.as_str())
            .unwrap_or_default();

        if formatstr.starts_with("%dyncols:") {
            let countryrecord = engine.country_info(country)?;
            add_dyncols(
                record,
                cityrecord,
                countryrecord,
                &nameslang,
                country,
                capital,
                column_values,
            );
            return Some(DYNCOLS_POPULATED.to_string());
        }

        return Some(format_result(
            engine, cityrecord, &nameslang, country, capital, formatstr, true,
        ));
    } else if mode == GeocodeSubCmd::Iplookup || mode == GeocodeSubCmd::IplookupNow {
        // check if the cell is an IP address
        let ip_addr = if let Ok(ip_addr) = cell.parse::<IpAddr>() {
            ip_addr
        } else {
            // not an IP address, check if it's a URL and lookup the IP address
            let url = Url::parse(cell)
                .map_err(|_| CliError::Other("Invalid URL".to_string()))
                .ok()?;
            let host = url.host_str().unwrap_or_default();
            // try to resolve the host to an IP address using cached DNS lookup
            cached_dns_lookup(host)?
        };

        let search_result = engine.geoip2_lookup(ip_addr);
        let Some(cityrecord) = search_result else {
            // if no cityrecord is found, return the IP address
            return Some(ip_addr.to_string());
        };
        let nameslang = get_cityrecord_name_in_lang(cityrecord, lang_lookup);
        let country = cityrecord.country.as_ref()?.code.as_str();
        let capital = engine
            .capital(country)
            .map(|cr| cr.name.as_str())
            .unwrap_or_default();
        #[allow(clippy::literal_string_with_formatting_args)]
        let formatstr = if formatstr == "%+" {
            if mode == GeocodeSubCmd::IplookupNow {
                "{name}, {admin1} {country}: {latitude}, {longitude}"
            } else {
                "%+"
            }
        } else {
            formatstr
        };

        // %dyncols: handling for IP lookup
        if formatstr.starts_with("%dyncols:") {
            let countryrecord = engine.country_info(country)?;
            add_dyncols(
                record,
                cityrecord,
                countryrecord,
                &nameslang,
                country,
                capital,
                column_values,
            );
            return Some(DYNCOLS_POPULATED.to_string());
        }

        return Some(format_result(
            engine, cityrecord, &nameslang, country, capital, formatstr, false,
        ));
    }

    // we're doing a Reverse/Now command and expect a WGS 84 coordinate
    // the regex validates for "(lat, long)" or "lat, long"
    // note that it is not pinned to the start of the string, so it can be in the middle
    // of a string, e.g. "The location of the incident is 40.7128, -74.0060"
    let locregex = LOCATION_REGEX();

    let loccaps = locregex.captures(cell);
    if let Some(loccaps) = loccaps {
        let lat = loccaps[1].to_string().parse::<f32>().unwrap_or_default();
        let long = loccaps[2].to_string().parse::<f32>().unwrap_or_default();
        if (-90.0..=90.0).contains(&lat) && (-180.0..=180.0).contains(&long) {
            let search_result = engine.reverse((lat, long), 1, k, country_filter_list);
            let cityrecord = {
                let search_result = search_result?;
                search_result.into_iter().next().map(|ri| ri.city)
            }?;

            let nameslang = get_cityrecord_name_in_lang(cityrecord, lang_lookup);

            let country = cityrecord.country.as_ref()?.code.as_str();

            if formatstr == "%+" {
                // default for reverse is city, admin1 country - e.g. "Brooklyn, New York US"
                return Some(format!(
                    "{cityname}, {admin1name} {country}",
                    cityname = nameslang.cityname,
                    admin1name = nameslang.admin1name,
                    country = country,
                ));
            }

            let capital = engine
                .capital(country)
                .map(|cr| cr.name.as_ref())
                .unwrap_or_default();

            if formatstr.starts_with("%dyncols:") {
                let countryrecord = engine.country_info(country)?;
                add_dyncols(
                    record,
                    cityrecord,
                    countryrecord,
                    &nameslang,
                    country,
                    capital,
                    column_values,
                );
                return Some(DYNCOLS_POPULATED.to_string());
            }

            return Some(format_result(
                engine, cityrecord, &nameslang, country, capital, formatstr, false,
            ));
        }
    }

    // not a valid lat, long
    None
}

/// Resolve a hostname to an IP address, cached so repeated hosts (common when an
/// IP-lookup column has many rows pointing at the same host) skip the blocking DNS
/// query. Like `search_index`, this uses a *sharded* concurrent LRU
/// (`#[concurrent_cached]` / `ShardedLruCache`) so the per-hit recency-bump write lock
/// doesn't serialize geocode's rayon-parallel pipeline; the shard count is auto-derived
/// from `available_parallelism()`. An `Option<IpAddr>` return skips caching `None`, so a
/// failed lookup is retried per-record rather than cached for the run (correct for
/// transient DNS failures). `max_size` must be an integer literal (~2M entries).
#[concurrent_cached(
    max_size = 2_000_000,
    key = "String",
    convert = r#"{ host.to_owned() }"#
)]
fn cached_dns_lookup(host: &str) -> Option<IpAddr> {
    dns_lookup::lookup_host(host)
        .map(|ips| {
            ips.into_iter()
                .next()
                .unwrap_or(IpAddr::V4(Ipv4Addr::UNSPECIFIED))
        })
        .ok()
}

/// "%dyncols:" formatstr used. Adds dynamic columns to CSV.
fn add_dyncols(
    record: &mut csv::StringRecord,
    cityrecord: &CitiesRecord,
    countryrecord: &CountryRecord,
    nameslang: &NamesLang,
    country: &str,
    capital: &str,
    column_values: &[&str],
) {
    for column in column_values {
        match *column {
            // CityRecord fields
            "id" => record.push_field(&cityrecord.id.to_string()),
            "name" => record.push_field(&nameslang.cityname),
            "latitude" => record.push_field(&cityrecord.latitude.to_string()),
            "longitude" => record.push_field(&cityrecord.longitude.to_string()),
            "country" => record.push_field(country),
            "admin1" => record.push_field(&nameslang.admin1name),
            "admin2" => record.push_field(&nameslang.admin2name),
            "capital" => record.push_field(capital),
            "timezone" => record.push_field(&cityrecord.timezone),
            "population" => record.push_field(&cityrecord.population.to_string()),

            // US FIPS fields
            "us_state_fips_code" => {
                let us_state_code = if let Some(admin1) = cityrecord.admin_division.as_ref() {
                    // admin1 code is a US state code, the two-letter state code
                    // is the last two characters of the admin1 code
                    // e.g. US.NY -> NY
                    // if not a US state code, return empty string
                    admin1.code.strip_prefix("US.").unwrap_or_default()
                } else {
                    // no admin1 code
                    // set to empty string
                    ""
                };
                // lookup US state FIPS code
                record.push_field(lookup_us_state_fips_code(us_state_code).unwrap_or_default());
            },
            "us_county_fips_code" => {
                let us_county_fips_code = if let Some(admin2) = cityrecord.admin2_division.as_ref()
                {
                    if admin2.code.starts_with("US.") && admin2.code.len() == 9 {
                        // admin2 code is a US county code, the three-digit county code
                        // is the last three characters of the admin2 code
                        // start at index 7 to skip the US. prefix
                        // e.g. US.NY.061 -> 061
                        format!("{:0>3}", &admin2.code[7..])
                    } else {
                        // admin2 code is not a US county code
                        // set to empty string
                        String::new()
                    }
                } else {
                    // no admin2 code
                    // set to empty string
                    String::new()
                };
                record.push_field(&us_county_fips_code);
            },

            // CountryRecord fields
            "country_name" => record.push_field(&nameslang.countryname),
            "iso3" => record.push_field(&countryrecord.info.iso3),
            "fips" => record.push_field(&countryrecord.info.fips),
            "area" => record.push_field(&countryrecord.info.area),
            "country_population" => record.push_field(&countryrecord.info.population.to_string()),
            "continent" => record.push_field(&countryrecord.info.continent),
            "tld" => record.push_field(&countryrecord.info.tld),
            "currency_code" => record.push_field(&countryrecord.info.currency_code),
            "currency_name" => record.push_field(&countryrecord.info.currency_name),
            "phone" => record.push_field(&countryrecord.info.phone),
            "postal_code_format" => record.push_field(&countryrecord.info.postal_code_format),
            "postal_code_regex" => record.push_field(&countryrecord.info.postal_code_regex),
            "languages" => record.push_field(&countryrecord.info.languages),
            "country_geonameid" => record.push_field(&countryrecord.info.geonameid.to_string()),
            "neighbours" => record.push_field(&countryrecord.info.neighbours),
            "equivalent_fips_code" => record.push_field(&countryrecord.info.equivalent_fips_code),

            // this should not happen as column_values has been pre-validated for these values
            _ => unreachable!(),
        }
    }
}

/// format the geocoded result based on formatstr if its not %+
#[cached(
    key = "String",
    convert = r#"{ format!("{}-{}-{}", cityrecord.id, formatstr, suggest_mode) }"#
)]
fn format_result(
    engine: &Engine,
    cityrecord: &CitiesRecord,
    nameslang: &NamesLang,
    country: &str,
    capital: &str,
    formatstr: &str,
    suggest_mode: bool,
) -> String {
    if formatstr.starts_with('%') {
        // if formatstr starts with %, then we're using a predefined format
        match formatstr {
            "%city-state" | "%city-admin1" => {
                format!("{}, {}", nameslang.cityname, nameslang.admin1name)
            },
            "%location" => format!("({}, {})", cityrecord.latitude, cityrecord.longitude),
            "%city-state-country" | "%city-admin1-country" => {
                format!(
                    "{}, {} {}",
                    nameslang.cityname, nameslang.admin1name, country
                )
            },
            "%lat-long" => format!("{}, {}", cityrecord.latitude, cityrecord.longitude),
            "%city-country" => format!("{}, {}", nameslang.cityname, country),
            "%city" => nameslang.cityname.clone(),
            "%city-county-state" | "%city-admin2-admin1" => {
                format!(
                    "{}, {}, {}",
                    nameslang.cityname, nameslang.admin2name, nameslang.admin1name,
                )
            },
            "%state" | "%admin1" => nameslang.admin1name.clone(),
            "%county" | "%admin2" => nameslang.admin2name.clone(),
            "%country" => country.to_string(),
            "%country_name" => nameslang.countryname.clone(),
            "%id" => cityrecord.id.to_string(),
            "%capital" => capital.to_string(),
            "%population" => cityrecord.population.to_string(),
            "%timezone" => cityrecord.timezone.to_string(),
            "%cityrecord" => format!("{cityrecord:?}"),
            "%admin1record" => format!("{:?}", cityrecord.admin_division),
            "%admin2record" => format!("{:?}", cityrecord.admin2_division),
            "%json" => {
                let Some(countryrecord) = engine.country_info(country) else {
                    return INVALID_COUNTRY_CODE.to_string();
                };
                let cr_json =
                    simd_json::to_string(cityrecord).unwrap_or_else(|_| "null".to_string());
                let country_json =
                    simd_json::to_string(countryrecord).unwrap_or_else(|_| "null".to_string());
                let us_fips_codes_json = get_us_fips_codes(cityrecord, nameslang);
                format!(
                    "{{\"cityrecord\":{cr_json}, \"countryrecord\":{country_json}, \
                     \"us_fips_codes\":{us_fips_codes_json}}}",
                )
            },
            "%pretty-json" => {
                let Some(countryrecord) = engine.country_info(country) else {
                    return INVALID_COUNTRY_CODE.to_string();
                };
                let cr_json =
                    simd_json::to_string_pretty(cityrecord).unwrap_or_else(|_| "null".to_string());
                let country_json = simd_json::to_string_pretty(countryrecord)
                    .unwrap_or_else(|_| "null".to_string());
                let us_fips_codes = get_us_fips_codes(cityrecord, nameslang);
                let us_fips_codes_json = simd_json::to_string_pretty(&us_fips_codes)
                    .unwrap_or_else(|_| "null".to_string());
                format!(
                    "{{\n  \"cityrecord\":{cr_json},\n  \"countryrecord\":{country_json},\n  \
                     \"us_fips_codes\":{us_fips_codes_json}\n}}",
                )
            },
            _ => {
                // invalid formatstr, so we use the default for suggest/now or reverse/now
                if suggest_mode {
                    // default for suggest is location - e.g. "(lat, long)"
                    format!(
                        "({latitude}, {longitude})",
                        latitude = cityrecord.latitude,
                        longitude = cityrecord.longitude
                    )
                } else {
                    // default for reverse/now or iplookup is city-admin1-country - e.g. "Brooklyn,
                    // New York US"
                    format!(
                        "{city}, {admin1} {country}",
                        city = nameslang.cityname,
                        admin1 = nameslang.admin1name,
                    )
                }
            },
        }
    } else {
        // if formatstr does not start with %, then we're using dynfmt,
        // i.e. twenty-eight predefined fields below in curly braces are replaced with values
        // e.g. "City: {name}, State: {admin1}, Country: {country} - {continent}"
        // unlike the predefined formats, we don't have a default format for dynfmt
        // so we return INVALID_DYNFMT if dynfmt fails to format the string
        // also, we have access to the country info fields as well

        // check if we have a valid country record
        let Some(countryrecord) = engine.country_info(country) else {
            return INVALID_COUNTRY_CODE.to_string();
        };

        // Now, parse the formatstr to get the fields to initialize in
        // the hashmap lookup. We do this so we only populate the hashmap with fields
        // that are actually used in the formatstr.
        let mut dynfmt_fields = Vec::with_capacity(10); // 10 is a reasonable default to save allocs
        let formatstr_re = FORMATSTR_REGEX();
        for format_fields in formatstr_re.captures_iter(formatstr) {
            // safety: the regex will always have a "key" group per the regex above
            dynfmt_fields.push(format_fields.name("key").unwrap().as_str());
        }

        let mut cityrecord_map: HashMap<&str, String> = HashMap::with_capacity(dynfmt_fields.len());

        for field in &dynfmt_fields {
            match *field {
                // cityrecord fields
                "id" => cityrecord_map.insert("id", cityrecord.id.to_string()),
                "name" => cityrecord_map.insert("name", nameslang.cityname.clone()),
                "latitude" => cityrecord_map.insert("latitude", cityrecord.latitude.to_string()),
                "longitude" => cityrecord_map.insert("longitude", cityrecord.longitude.to_string()),
                "country" => cityrecord_map.insert("country", country.to_string()),
                "country_name" => {
                    cityrecord_map.insert("country_name", nameslang.countryname.clone())
                },
                "admin1" => cityrecord_map.insert("admin1", nameslang.admin1name.clone()),
                "admin2" => cityrecord_map.insert("admin2", nameslang.admin2name.clone()),
                "capital" => cityrecord_map.insert("capital", capital.to_string()),
                "timezone" => cityrecord_map.insert("timezone", cityrecord.timezone.to_string()),
                "population" => {
                    cityrecord_map.insert("population", cityrecord.population.to_string())
                },

                // US FIPS fields
                // set US state FIPS code
                "us_state_fips_code" => {
                    let us_state_code = if let Some(admin1) = cityrecord.admin_division.as_ref() {
                        admin1.code.strip_prefix("US.").unwrap_or_default()
                    } else {
                        ""
                    };
                    cityrecord_map.insert(
                        "us_state_fips_code",
                        lookup_us_state_fips_code(us_state_code)
                            .unwrap_or("")
                            .to_string(),
                    )
                },

                // set US county FIPS code
                "us_county_fips_code" => cityrecord_map.insert("us_county_fips_code", {
                    match cityrecord.admin2_division.as_ref() {
                        Some(admin2) => {
                            if admin2.code.starts_with("US.") && admin2.code.len() == 9 {
                                // admin2 code is a US county code, the three-digit county code
                                // is the last three characters of the admin2 code
                                // start at index 7 to skip the US. prefix
                                // e.g. US.NY.061 -> 061
                                format!("{:0>3}", &admin2.code[7..])
                            } else {
                                // admin2 code is not a US county code
                                // set to empty string
                                String::new()
                            }
                        },
                        None => {
                            // no admin2 code
                            // set to empty string
                            String::new()
                        },
                    }
                }),

                // countryrecord fields
                "iso3" => cityrecord_map.insert("iso3", countryrecord.info.iso3.to_string()),
                "fips" => cityrecord_map.insert("fips", countryrecord.info.fips.to_string()),
                "area" => cityrecord_map.insert("area", countryrecord.info.area.to_string()),
                "country_population" => cityrecord_map.insert(
                    "country_population",
                    countryrecord.info.population.to_string(),
                ),
                "continent" => {
                    cityrecord_map.insert("continent", countryrecord.info.continent.to_string())
                },
                "tld" => cityrecord_map.insert("tld", countryrecord.info.tld.to_string()),
                "currency_code" => cityrecord_map.insert(
                    "currency_code",
                    countryrecord.info.currency_code.to_string(),
                ),
                "currency_name" => cityrecord_map.insert(
                    "currency_name",
                    countryrecord.info.currency_name.to_string(),
                ),
                "phone" => cityrecord_map.insert("phone", countryrecord.info.phone.to_string()),
                "postal_code_format" => cityrecord_map.insert(
                    "postal_code_format",
                    countryrecord.info.postal_code_format.to_string(),
                ),
                "postal_code_regex" => cityrecord_map.insert(
                    "postal_code_regex",
                    countryrecord.info.postal_code_regex.to_string(),
                ),
                "languages" => {
                    cityrecord_map.insert("languages", countryrecord.info.languages.to_string())
                },
                "country_geonameid" => cityrecord_map.insert(
                    "country_geonameid",
                    countryrecord.info.geonameid.to_string(),
                ),
                "neighbours" => {
                    cityrecord_map.insert("neighbours", countryrecord.info.neighbours.to_string())
                },
                "equivalent_fips_code" => cityrecord_map.insert(
                    "equivalent_fips_code",
                    countryrecord.info.equivalent_fips_code.to_string(),
                ),
                _ => return INVALID_DYNFMT.to_string(),
            };
        }

        if let Ok(formatted) = dynfmt2::SimpleCurlyFormat.format(formatstr, cityrecord_map) {
            formatted.to_string()
        } else {
            INVALID_DYNFMT.to_string()
        }
    }
}

/// Helper: return the first present `OpenCage` component (as a plain string) from `keys`.
fn opencage_component(r: &OpencageResult, keys: &[&str]) -> String {
    for key in keys {
        if let Some(value) = r.components.get(*key) {
            return opencage_value_to_string(value);
        }
    }
    String::new()
}

/// Helper: convert a `serde_json::Value` to a plain (unquoted) string.
fn opencage_value_to_string(value: &serde_json::Value) -> String {
    match value {
        serde_json::Value::String(s) => s.clone(),
        serde_json::Value::Null => String::new(),
        other => other.to_string(),
    }
}

/// Helper: walk a `serde_json::Value` by a dotted path (e.g. "timezone.name").
fn opencage_json_dotpath(value: &serde_json::Value, path: &str) -> String {
    let mut current = value;
    for part in path.split('.') {
        match current.get(part) {
            Some(next) => current = next,
            None => return String::new(),
        }
    }
    opencage_value_to_string(current)
}

/// Helper: resolve a single dynamic-format field for an `OpenCage` result.
/// Returns None for an unrecognized key (which makes the whole template invalid).
fn opencage_field_value(r: &OpencageResult, key: &str) -> Option<String> {
    match key {
        "formatted" => Some(r.formatted.clone()),
        "lat" => Some(r.geometry.lat.to_string()),
        "lng" => Some(r.geometry.lng.to_string()),
        "confidence" => Some(r.confidence.to_string()),
        _ => {
            if let Some(name) = key.strip_prefix("components.") {
                Some(
                    r.components
                        .get(name)
                        .map(opencage_value_to_string)
                        .unwrap_or_default(),
                )
            } else {
                key.strip_prefix("annotations.")
                    .map(|path| opencage_json_dotpath(&r.annotations, path))
            }
        },
    }
}

/// Returns true if `key` is a valid "%dyncols:" field key for an `OpenCage` result,
/// i.e. one that `opencage_field_value()` can resolve.
/// `components.` / `annotations.` prefixes require a non-empty suffix - a bare
/// prefix has no field to resolve and would silently yield an empty column.
fn is_valid_opencage_dyncol(key: &str) -> bool {
    matches!(key, "formatted" | "lat" | "lng" | "confidence")
        || key
            .strip_prefix("components.")
            .or_else(|| key.strip_prefix("annotations."))
            .is_some_and(|suffix| !suffix.is_empty())
}

/// Build a JSON object from an `OpenCage` result for the %json / %pretty-json formats.
fn opencage_result_json(r: &OpencageResult, no_annotations: bool) -> serde_json::Value {
    let mut obj = json!({
        "formatted": r.formatted,
        "confidence": r.confidence,
        "geometry": {"lat": r.geometry.lat, "lng": r.geometry.lng},
        "components": r.components,
    });
    if !no_annotations && !r.annotations.is_null() {
        obj["annotations"] = r.annotations.clone();
    }
    obj
}

/// Format an `OpenCage` geocoding result per `formatstr`.
/// Supports predefined %-formats and dotted-key dynamic templating.
/// Unlike `format_result()`, this is bound to the `OpenCage` response shape, not Geonames.
fn format_opencage_result(r: &OpencageResult, formatstr: &str, no_annotations: bool) -> String {
    if formatstr.starts_with('%') {
        #[allow(clippy::match_same_arms)]
        match formatstr {
            "%+" | "%formatted" => r.formatted.clone(),
            "%lat-long" => format!("{}, {}", r.geometry.lat, r.geometry.lng),
            "%location" => format!("({}, {})", r.geometry.lat, r.geometry.lng),
            "%city" => opencage_component(
                r,
                &[
                    "city",
                    "town",
                    "village",
                    "municipality",
                    "hamlet",
                    "locality",
                ],
            ),
            "%state" | "%admin1" => opencage_component(r, &["state", "province", "region"]),
            "%county" | "%admin2" => {
                opencage_component(r, &["county", "state_district", "district"])
            },
            "%country" => opencage_component(r, &["country_code"]).to_uppercase(),
            "%country_name" => opencage_component(r, &["country"]),
            "%postcode" => opencage_component(r, &["postcode"]),
            "%confidence" => r.confidence.to_string(),
            "%json" => serde_json::to_string(&opencage_result_json(r, no_annotations))
                .unwrap_or_else(|_| "null".to_string()),
            "%pretty-json" => {
                serde_json::to_string_pretty(&opencage_result_json(r, no_annotations))
                    .unwrap_or_else(|_| "null".to_string())
            },
            // unknown %-format, fall back to the default (the formatted address)
            _ => r.formatted.clone(),
        }
    } else {
        // dynamic templating with dotted keys, e.g.
        // "{components.city}, {annotations.timezone.name}"
        let re = OPENCAGE_FORMATSTR_REGEX();
        let mut fields: HashMap<String, String> = HashMap::new();
        for caps in re.captures_iter(formatstr) {
            let key = caps[1].to_string();
            if fields.contains_key(&key) {
                continue;
            }
            let Some(value) = opencage_field_value(r, &key) else {
                return INVALID_DYNFMT.to_string();
            };
            fields.insert(key, value);
        }
        re.replace_all(formatstr, |caps: &regex::Captures| {
            fields.get(&caps[1]).cloned().unwrap_or_default()
        })
        .into_owned()
    }
}

/// Normalize a query cell for the `OpenCage` API.
/// If the cell (after stripping one pair of surrounding parens) is entirely a
/// WGS-84 coordinate, returns a bare "lat,long" string for reverse geocoding.
/// Otherwise, for forward geocoding, returns the trimmed cell as-is.
/// When `reverse` is set, a non-coordinate cell returns None (an invalid row).
fn normalize_opencage_query(cell: &str, reverse: bool) -> Option<String> {
    let trimmed = cell.trim();
    let mut coord_candidate = trimmed;
    if coord_candidate.starts_with('(') && coord_candidate.ends_with(')') {
        coord_candidate = coord_candidate[1..coord_candidate.len() - 1].trim();
    }
    if let Some(caps) = LOCATION_REGEX().captures(coord_candidate)
        && let Some(m) = caps.get(0)
        && m.start() == 0
        && m.end() == coord_candidate.len()
    {
        // the whole cell is a coordinate -> normalize to bare "lat,long"
        return Some(format!("{},{}", &caps[1], &caps[2]));
    }
    if reverse {
        // --reverse forces coordinate input; a non-coordinate cell is invalid
        None
    } else {
        // forward geocoding: pass the address as-is
        Some(trimmed.to_string())
    }
}

/// Encode an `OpenCage` lookup outcome for the string-valued cache.
/// `Some(s)` is stored as "F"+s; `None` (zero results) is stored as "N".
fn encode_opencage_cache(outcome: Option<&str>) -> String {
    match outcome {
        Some(s) => format!("F{s}"),
        None => "N".to_string(),
    }
}

/// Decode a cached `OpenCage` lookup outcome (see `encode_opencage_cache`).
fn decode_opencage_cache(encoded: &str) -> Option<String> {
    if encoded == "N" {
        None
    } else {
        Some(encoded.get(1..).unwrap_or_default().to_string())
    }
}

/// Call the `OpenCage` Geocoding API for a single (already normalized) query.
/// The same `q` parameter does forward geocoding for an address and reverse
/// geocoding for a "lat,long" coordinate.
async fn opencage_fetch(
    client: &reqwest::Client,
    limiter: &DefaultDirectRateLimiter,
    api_key: &str,
    query: &str,
    language: &str,
    countrycode: Option<&str>,
    no_annotations: bool,
) -> Result<OpencageResponse, OcError> {
    // respect the configured rate limit before hitting the API
    limiter.until_ready().await;

    let no_ann = if no_annotations { "1" } else { "0" };
    let cc_lower = countrycode.map(str::to_lowercase);
    let mut params: Vec<(&str, &str)> = vec![
        ("q", query),
        ("key", api_key),
        ("limit", "1"),
        ("language", language),
        ("no_annotations", no_ann),
        ("abbrv", "0"),
    ];
    if let Some(cc) = &cc_lower {
        params.push(("countrycode", cc.as_str()));
    }

    let url = Url::parse_with_params("https://api.opencagedata.com/geocode/v1/json", &params)
        .map_err(|e| OcError::Transient(format!("OpenCage URL build error: {e}")))?;
    let resp = client
        .get(url)
        .send()
        .await
        .map_err(|e| OcError::Transient(format!("OpenCage network error: {e}")))?;

    let status = resp.status().as_u16();
    match status {
        200 => {},
        401 | 403 => {
            return Err(OcError::Fatal(format!(
                "OpenCage rejected the API key (HTTP {status}). Check --api-key or the \
                 QSV_OPENCAGE_API_KEY environment variable."
            )));
        },
        402 => {
            return Err(OcError::Fatal(
                "OpenCage quota exceeded (HTTP 402). The free tier allows 2,500 requests/day."
                    .to_string(),
            ));
        },
        429 => {
            return Err(OcError::Transient(
                "OpenCage rate-limited the request (HTTP 429).".to_string(),
            ));
        },
        code => {
            return Err(OcError::Transient(format!(
                "OpenCage returned HTTP {code}."
            )));
        },
    }

    let parsed: OpencageResponse = resp
        .json()
        .await
        .map_err(|e| OcError::Transient(format!("OpenCage response error: {e}")))?;

    // OpenCage also echoes the status code in the JSON body
    match parsed.status.code {
        401 | 403 => Err(OcError::Fatal(
            "OpenCage rejected the API key. Check --api-key or the QSV_OPENCAGE_API_KEY \
             environment variable."
                .to_string(),
        )),
        402 => Err(OcError::Fatal(
            "OpenCage quota exceeded. The free tier allows 2,500 requests/day.".to_string(),
        )),
        _ => Ok(parsed),
    }
}

/// Geocode a single normalized query via `OpenCage`, with caching.
/// Checks the in-run memory cache, then the persistent on-disk cache, and only
/// then hits the API. Successful lookups (including zero-result lookups) are
/// cached; `OcError::Fatal` aborts the run and is never cached.
#[allow(clippy::too_many_arguments)]
async fn opencage_lookup(
    client: &reqwest::Client,
    limiter: &DefaultDirectRateLimiter,
    disk_cache: Option<&RedbCache<String, String>>,
    mem_cache: &mut HashMap<String, String>,
    api_key: &str,
    query: &str,
    formatstr: &str,
    language: &str,
    countrycode: Option<&str>,
    no_annotations: bool,
) -> Result<Option<String>, OcError> {
    let cache_key = format!(
        "{query}|{language}|{cc}|{formatstr}|{no_annotations}",
        cc = countrycode.unwrap_or_default()
    );

    if let Some(encoded) = mem_cache.get(&cache_key) {
        return Ok(decode_opencage_cache(encoded));
    }
    if let Some(dc) = disk_cache
        && let Ok(Some(encoded)) = dc.cache_get(&cache_key)
    {
        let decoded = decode_opencage_cache(&encoded);
        mem_cache.insert(cache_key, encoded);
        return Ok(decoded);
    }

    let response = opencage_fetch(
        client,
        limiter,
        api_key,
        query,
        language,
        countrycode,
        no_annotations,
    )
    .await?;

    let outcome = response
        .results
        .first()
        .map(|r| format_opencage_result(r, formatstr, no_annotations));

    let encoded = encode_opencage_cache(outcome.as_deref());
    if let Some(dc) = disk_cache {
        let _ = dc.cache_set(cache_key.clone(), encoded.clone());
    }
    mem_cache.insert(cache_key, encoded);
    Ok(outcome)
}

/// Geocode a single normalized query via `OpenCage` for "%dyncols:" mode, with caching.
/// Unlike `opencage_lookup` (which caches a single formatted string), this caches the
/// raw first result as JSON, so any set of dyncols fields can be extracted from it -
/// the cache key is therefore independent of the requested columns.
/// Returns one value per `column_values` entry, or None for a zero-result lookup.
#[allow(clippy::too_many_arguments)]
async fn opencage_lookup_dyncols(
    client: &reqwest::Client,
    limiter: &DefaultDirectRateLimiter,
    disk_cache: Option<&RedbCache<String, String>>,
    mem_cache: &mut HashMap<String, String>,
    api_key: &str,
    query: &str,
    language: &str,
    countrycode: Option<&str>,
    no_annotations: bool,
    column_values: &[String],
) -> Result<Option<Vec<String>>, OcError> {
    // the literal NUL in the formatstr slot keeps this key space disjoint from
    // opencage_lookup's: a --formatstr is a CLI arg and can never contain a NUL
    // byte, so a dyncols cache entry can never collide with a regular one.
    let cache_key = format!(
        "{query}|{language}|{cc}|\0dyncols|{no_annotations}",
        cc = countrycode.unwrap_or_default()
    );

    // extract the requested column values from a cached raw-result JSON string
    let extract = |encoded: &str| -> Option<Vec<String>> {
        let result_json = decode_opencage_cache(encoded)?;
        let result: OpencageResult = serde_json::from_str(&result_json).ok()?;
        Some(
            column_values
                .iter()
                .map(|key| opencage_field_value(&result, key).unwrap_or_default())
                .collect(),
        )
    };

    if let Some(encoded) = mem_cache.get(&cache_key) {
        return Ok(extract(encoded));
    }
    if let Some(dc) = disk_cache
        && let Ok(Some(encoded)) = dc.cache_get(&cache_key)
    {
        let values = extract(&encoded);
        mem_cache.insert(cache_key, encoded);
        return Ok(values);
    }

    let response = opencage_fetch(
        client,
        limiter,
        api_key,
        query,
        language,
        countrycode,
        no_annotations,
    )
    .await?;

    // cache the raw first result as JSON (a zero-result lookup is cached too)
    let result_json = match response.results.first() {
        Some(r) => match serde_json::to_string(&opencage_result_json(r, no_annotations)) {
            Ok(json) => Some(json),
            Err(_) => {
                // serialization unexpectedly failed: skip caching rather than
                // poisoning the cache with an empty entry that would later
                // decode as a (bogus) zero-result. Extract the column values
                // directly from the live result instead.
                return Ok(Some(
                    column_values
                        .iter()
                        .map(|key| opencage_field_value(r, key).unwrap_or_default())
                        .collect(),
                ));
            },
        },
        None => None,
    };
    let encoded = encode_opencage_cache(result_json.as_deref());
    if let Some(dc) = disk_cache {
        let _ = dc.cache_set(cache_key.clone(), encoded.clone());
    }
    let values = extract(&encoded);
    mem_cache.insert(cache_key, encoded);
    Ok(values)
}

/// `get_countryinfo` is a cached function that returns a countryinfo result for a given cell value.
/// It is used by the countryinfo/countryinfonow subcommands.
#[cached(key = "String", convert = r#"{ format!("{cell}-{formatstr}") }"#)]
fn get_countryinfo(
    engine: &Engine,
    cell: &str,
    lang_lookup: &str,
    formatstr: &str,
) -> Option<String> {
    let Some(countryrecord) = engine.country_info(&cell.to_ascii_uppercase()) else {
        // no results, so return early with None
        return None;
    };

    if formatstr.starts_with('%') {
        // if formatstr starts with %, then we're using a predefined format
        let formatted = match formatstr {
            "%capital" => countryrecord.info.capital.to_string(),
            "%continent" => countryrecord.info.continent.to_string(),
            "%json" => simd_json::to_string(countryrecord).unwrap_or_else(|_| "null".to_string()),
            "%pretty-json" => {
                simd_json::to_string_pretty(countryrecord).unwrap_or_else(|_| "null".to_string())
            },
            _ => countryrecord
                .names
                .as_ref()
                .and_then(|n| n.get(lang_lookup))
                .map(ToString::to_string)
                .unwrap_or_default(),
        };
        Some(formatted)
    } else {
        // if formatstr does not start with %, then we're using dynfmt,
        // i.e. sixteen predefined fields below in curly braces are replaced with values
        // e.g. "Country: {country_name}, Continent: {continent} Currency: {currency_name}
        // ({currency_code})})"

        // first, parse the formatstr to get the fields to initialixe in the hashmap lookup
        // we do this so we only populate the hashmap with fields that are actually used
        // in the formatstr.
        let mut dynfmt_fields = Vec::with_capacity(10); // 10 is a reasonable default to save allocs
        let formatstr_re = FORMATSTR_REGEX();
        for format_fields in formatstr_re.captures_iter(formatstr) {
            // safety: the regex will always have a "key" group per the regex above
            dynfmt_fields.push(format_fields.name("key").unwrap().as_str());
        }

        let mut countryrecord_map: HashMap<&str, String> =
            HashMap::with_capacity(dynfmt_fields.len());

        for field in &dynfmt_fields {
            match *field {
                "country_name" => countryrecord_map.insert("country_name", {
                    countryrecord
                        .names
                        .as_ref()
                        .and_then(|n| n.get(lang_lookup))
                        .map(ToString::to_string)
                        .unwrap_or_default()
                }),
                "iso3" => countryrecord_map.insert("iso3", countryrecord.info.iso3.to_string()),
                "fips" => countryrecord_map.insert("fips", countryrecord.info.fips.to_string()),
                "capital" => {
                    countryrecord_map.insert("capital", countryrecord.info.capital.to_string())
                },
                "area" => countryrecord_map.insert("area", countryrecord.info.area.to_string()),
                "country_population" => countryrecord_map.insert(
                    "country_population",
                    countryrecord.info.population.to_string(),
                ),
                "continent" => {
                    countryrecord_map.insert("continent", countryrecord.info.continent.to_string())
                },
                "tld" => countryrecord_map.insert("tld", countryrecord.info.tld.to_string()),
                "currency_code" => countryrecord_map.insert(
                    "currency_code",
                    countryrecord.info.currency_code.to_string(),
                ),
                "currency_name" => countryrecord_map.insert(
                    "currency_name",
                    countryrecord.info.currency_name.to_string(),
                ),
                "phone" => countryrecord_map.insert("phone", countryrecord.info.phone.to_string()),
                "postal_code_format" => countryrecord_map.insert(
                    "postal_code_format",
                    countryrecord.info.postal_code_format.to_string(),
                ),
                "postal_code_regex" => countryrecord_map.insert(
                    "postal_code_regex",
                    countryrecord.info.postal_code_regex.to_string(),
                ),
                "languages" => {
                    countryrecord_map.insert("languages", countryrecord.info.languages.to_string())
                },
                "geonameid" => {
                    countryrecord_map.insert("geonameid", countryrecord.info.geonameid.to_string())
                },
                "neighbours" => countryrecord_map
                    .insert("neighbours", countryrecord.info.neighbours.to_string()),
                "equivalent_fips_code" => countryrecord_map.insert(
                    "equivalent_fips_code",
                    countryrecord.info.equivalent_fips_code.to_string(),
                ),
                _ => return Some(INVALID_DYNFMT.to_string()),
            };
        }

        if let Ok(formatted) = dynfmt2::SimpleCurlyFormat.format(formatstr, countryrecord_map) {
            Some(formatted.to_string())
        } else {
            Some(INVALID_DYNFMT.to_string())
        }
    }
}

/// `get_cityrecord_name_in_lang` is a cached function that returns a `NamesLang` struct
/// containing the city, admin1, admin2, and country names in the specified language.
/// Note that the index file needs to be built with the desired languages for this to work.
/// Use the "index-update" subcommand with the --languages option to rebuild the index
/// with the desired languages. Otherwise, all names will be in English (en)
#[cached(
    key = "String",
    convert = r#"{ format!("{}-{lang_lookup}", cityrecord.id) }"#
)]
fn get_cityrecord_name_in_lang(cityrecord: &CitiesRecord, lang_lookup: &str) -> NamesLang {
    let cityname = cityrecord
        .names
        .as_ref()
        .and_then(|n| n.get(lang_lookup))
        // Note that the city name is the default name if the language is not found.
        .unwrap_or(&cityrecord.name)
        .to_string();
    let admin1name = cityrecord
        .admin1_names
        .as_ref()
        .and_then(|n| n.get(lang_lookup))
        .map(ToString::to_string)
        .unwrap_or_default();
    let admin2name = cityrecord
        .admin2_names
        .as_ref()
        .and_then(|n| n.get(lang_lookup))
        .map(ToString::to_string)
        .unwrap_or_default();
    let countryname = cityrecord
        .country_names
        .as_ref()
        .and_then(|n| n.get(lang_lookup))
        .map(ToString::to_string)
        .unwrap_or_default();

    NamesLang {
        cityname,
        admin1name,
        admin2name,
        countryname,
    }
}

/// Resolve the geocode cache directory, honoring the `QSV_CACHE_DIR` env var (which takes
/// precedence) and falling back to `fallback_cache_dir` (e.g. the `--cache-dir` flag default
/// of `~/.qsv-cache`). Leading `~` is expanded. The directory is created if it doesn't exist.
fn resolve_geocode_cache_dir(fallback_cache_dir: &str) -> CliResult<PathBuf> {
    let cache_dir = if let Ok(env_cache_dir) = std::env::var("QSV_CACHE_DIR") {
        if env_cache_dir.starts_with('~') {
            // QSV_CACHE_DIR starts with ~, expand it; fall back to literal path if HOME unset
            expand_tilde(&env_cache_dir).unwrap_or_else(|| PathBuf::from(&env_cache_dir))
        } else {
            PathBuf::from(env_cache_dir)
        }
    } else if fallback_cache_dir.starts_with('~') {
        expand_tilde(fallback_cache_dir).unwrap_or_else(|| PathBuf::from(fallback_cache_dir))
    } else {
        PathBuf::from(fallback_cache_dir)
    };
    if !Path::new(&cache_dir).exists() {
        fs::create_dir_all(&cache_dir)?;
    }
    Ok(cache_dir)
}

/// Structured reverse-geocoding result for a single coordinate, used by callers outside
/// `geocode` (e.g. `viz smart`) that need the location components rather than a preformatted
/// string. `country` is the localized country name when available, otherwise the 2-letter code.
///
/// The trailing fields are `viz`-only enrichment (empty/zero when unavailable): `admin2` is the
/// localized county name; `country_iso2` is the ISO-3166-1 alpha-2 code (for a follow-up
/// [`country_context`] lookup); `us_state_fips`/`us_county_fips` are the 2-digit state and 5-digit
/// combined county FIPS codes for US matches. `Default` lets test literals spread `..`.
#[derive(Clone, Debug, Default)]
pub struct GeoLabel {
    pub city:           String,
    pub admin1:         String,
    pub country:        String,
    pub admin2:         String,
    pub country_iso2:   String,
    pub us_state_fips:  String,
    pub us_county_fips: String,
}

/// Reverse-geocode a batch of `(lat, lon)` points against the local Geonames index, loading
/// the engine ONCE. Returns one `Option<GeoLabel>` per input point (in order); `None` where no
/// city matched (e.g. a coordinate over open water). Returns `Err` only on a hard setup failure
/// (tokio runtime creation, or index download/load failure, e.g. offline on first use) — callers
/// like `viz smart` treat `Err` as "no metadata" and degrade gracefully.
///
/// `lang` selects the language for the returned names (defaults to `"en"`); note the index must
/// have been built with that language for non-English names to resolve.
#[allow(clippy::cast_possible_truncation)]
pub fn reverse_geocode_points(
    points: &[(f64, f64)],
    lang: Option<&str>,
) -> CliResult<Vec<Option<GeoLabel>>> {
    let lang_lookup = lang.unwrap_or("en");

    // resolve the active index file path the same way geocode_main does, so viz shares the
    // exact same on-disk Geonames index (default `~/.qsv-cache`).
    let geocode_cache_dir = resolve_geocode_cache_dir("~/.qsv-cache")?;
    let geocode_index_filename = std::env::var("QSV_GEOCODE_INDEX_FILENAME")
        .unwrap_or_else(|_| DEFAULT_GEOCODE_INDEX_FILENAME.to_string());
    let geocode_index_file = PathBuf::from(format!(
        "{}/{geocode_index_filename}",
        geocode_cache_dir.display()
    ));

    // geosuggest's index loader is async (it may download the index on first use); spin up a
    // tokio runtime to drive it, mirroring geocode::run. The reverse lookups themselves are sync.
    let rt = tokio::runtime::Runtime::new()?;
    let progressbar = ProgressBar::hidden();
    let engine_data = rt.block_on(load_engine_data(geocode_index_file, &progressbar))?;
    // `as_engine` borrows `engine_data`, so it must outlive `engine` and all lookups below.
    let engine = engine_data
        .as_engine()
        .map_err(|e| CliError::Other(format!("Error initializing geocode Engine: {e}")))?;

    let mut results = Vec::with_capacity(points.len());
    for &(lat, lon) in points {
        let label = (|| {
            if !((-90.0..=90.0).contains(&lat) && (-180.0..=180.0).contains(&lon)) {
                return None;
            }
            // turbofish pins the unused `countries` filter's element type (both filters are None).
            let search_result = engine.reverse::<String>((lat as f32, lon as f32), 1, None, None);
            let cityrecord = search_result?.into_iter().next().map(|ri| ri.city)?;
            let nameslang = get_cityrecord_name_in_lang(cityrecord, lang_lookup);
            // prefer the localized country name; fall back to the 2-letter country code.
            let country = if nameslang.countryname.is_empty() {
                cityrecord
                    .country
                    .as_ref()
                    .map(|c| c.code.as_str().to_string())
                    .unwrap_or_default()
            } else {
                nameslang.countryname.clone()
            };
            let (us_state_fips, us_county_fips) = us_fips_strings(cityrecord);
            Some(GeoLabel {
                city: nameslang.cityname.clone(),
                admin1: nameslang.admin1name.clone(),
                country,
                admin2: nameslang.admin2name.clone(),
                country_iso2: cityrecord
                    .country
                    .as_ref()
                    .map(|c| c.code.as_str().to_string())
                    .unwrap_or_default(),
                us_state_fips,
                us_county_fips,
            })
        })();
        results.push(label);
    }

    Ok(results)
}

/// Structured region identifiers for a single geocoded location, used by callers outside
/// `geocode` (e.g. `viz choropleth`) that need region CODES — ISO-3 country and US state codes —
/// to drive a choropleth, which the name-only [`GeoLabel`] does not expose.
///
/// This is a complete region-identifier surface; current callers read only `iso3`/`us_state_code`,
/// but the country/admin name and admin1 code fields are populated for reuse by future callers.
#[allow(dead_code)]
#[derive(Clone, Debug)]
pub struct GeoRegion {
    /// ISO-3166-1 alpha-2 country code, e.g. "US".
    pub iso2:          String,
    /// ISO-3166-1 alpha-3 country code, e.g. "USA". Empty if the country record is unavailable.
    pub iso3:          String,
    /// Localized country name when available, otherwise the 2-letter code.
    pub country_name:  String,
    /// Localized admin1 (state/province) name.
    pub admin1_name:   String,
    /// admin1 code in Geonames format, e.g. "US.NY"; `None` if the record has no admin1.
    pub admin1_code:   Option<String>,
    /// 2-letter US state code (e.g. "NY"), `Some` only when the country is the US.
    pub us_state_code: Option<String>,
}

/// Load the shared Geonames engine (downloading the index on first use), run `f` against it, and
/// return its owned result. Centralizes the index-resolution + tokio + `as_engine` boilerplate so
/// the engine (which borrows `EngineData`) stays alive for the whole batch and is dropped after
/// `f` returns. Returns `Err` only on a hard setup failure (tokio runtime, index download/load).
fn with_geocode_engine<F, R>(f: F) -> CliResult<R>
where
    F: FnOnce(&Engine) -> R,
{
    let geocode_cache_dir = resolve_geocode_cache_dir("~/.qsv-cache")?;
    let geocode_index_filename = std::env::var("QSV_GEOCODE_INDEX_FILENAME")
        .unwrap_or_else(|_| DEFAULT_GEOCODE_INDEX_FILENAME.to_string());
    let geocode_index_file = PathBuf::from(format!(
        "{}/{geocode_index_filename}",
        geocode_cache_dir.display()
    ));
    let rt = tokio::runtime::Runtime::new()?;
    let progressbar = ProgressBar::hidden();
    let engine_data = rt.block_on(load_engine_data(geocode_index_file, &progressbar))?;
    let engine = engine_data
        .as_engine()
        .map_err(|e| CliError::Other(format!("Error initializing geocode Engine: {e}")))?;
    Ok(f(&engine))
}

/// Derive a [`GeoRegion`] from a matched `CitiesRecord`. Returns `None` when the record has no
/// country (the minimum needed for a region code). `iso3` is resolved via a country-info lookup;
/// `us_state_code` is the 2-letter state code only when the country is the US.
fn cityrecord_to_region(
    engine: &Engine,
    cityrecord: &CitiesRecord,
    lang_lookup: &str,
) -> Option<GeoRegion> {
    let iso2 = cityrecord.country.as_ref()?.code.as_str().to_string();
    let nameslang = get_cityrecord_name_in_lang(cityrecord, lang_lookup);
    let country_name = if nameslang.countryname.is_empty() {
        iso2.clone()
    } else {
        nameslang.countryname.clone()
    };
    let iso3 = engine
        .country_info(&iso2)
        .map(|cr| cr.info.iso3.to_string())
        .unwrap_or_default();
    let admin1_code = cityrecord
        .admin_division
        .as_ref()
        .map(|a| a.code.to_string());
    let us_state_code = if iso2 == "US" {
        admin1_code
            .as_deref()
            .and_then(|c| c.strip_prefix("US."))
            .map(str::to_string)
    } else {
        None
    };
    Some(GeoRegion {
        iso2,
        iso3,
        country_name,
        admin1_name: nameslang.admin1name.clone(),
        admin1_code,
        us_state_code,
    })
}

/// Reverse-geocode a batch of `(lat, lon)` points to region codes (see [`GeoRegion`]), loading the
/// engine ONCE. Returns one `Option<GeoRegion>` per input point (in order); `None` where no city
/// matched or the match lacks a country. `Err` only on a hard setup failure — callers degrade
/// gracefully, exactly like [`reverse_geocode_points`]. `lang` selects the localized-name language
/// (defaults to `"en"`).
#[allow(clippy::cast_possible_truncation)]
pub fn reverse_geocode_regions(
    points: &[(f64, f64)],
    lang: Option<&str>,
) -> CliResult<Vec<Option<GeoRegion>>> {
    let lang_lookup = lang.unwrap_or("en");
    with_geocode_engine(|engine| {
        points
            .iter()
            .map(|&(lat, lon)| {
                if !((-90.0..=90.0).contains(&lat) && (-180.0..=180.0).contains(&lon)) {
                    return None;
                }
                // turbofish pins the unused `countries` filter's element type (both filters None).
                let search_result =
                    engine.reverse::<String>((lat as f32, lon as f32), 1, None, None);
                let cityrecord = search_result?.into_iter().next().map(|ri| ri.city)?;
                cityrecord_to_region(engine, cityrecord, lang_lookup)
            })
            .collect()
    })
}

/// Forward-geocode a batch of place names to region codes (see [`GeoRegion`]) via the same fuzzy
/// `suggest` engine the `geocode suggest` subcommand uses, loading the engine ONCE. Returns one
/// `Option<GeoRegion>` per input name (in order); `None` where no city matched or the match lacks
/// a country. `Err` only on a hard setup failure. `lang` selects the localized-name language.
pub fn forward_geocode_regions(
    names: &[&str],
    lang: Option<&str>,
) -> CliResult<Vec<Option<GeoRegion>>> {
    let lang_lookup = lang.unwrap_or("en");
    with_geocode_engine(|engine| {
        names
            .iter()
            .map(|name| {
                // turbofish pins the unused `countries` filter's element type.
                let cityrecord = engine
                    .suggest::<String>(name, 1, None, None)
                    .into_iter()
                    .next()?;
                cityrecord_to_region(engine, cityrecord, lang_lookup)
            })
            .collect()
    })
}

/// Per-country context for a batch of ISO-3166-1 alpha-2 codes, used by `viz smart` to annotate a
/// map's spatial-extent summary. `continent` is the Geonames continent name (`""` when
/// unavailable).
#[derive(Clone, Debug, Default)]
pub struct CountryContext {
    pub continent: String,
}

/// Resolve [`CountryContext`] for each ISO-2 code (in order), loading the engine ONCE. Returns
/// `None` for a code that doesn't resolve (or carries no continent). `Err` only on a hard setup
/// failure — callers (like `viz smart`) treat `Err` as "no context" and degrade gracefully.
/// `continent` is expanded to a human-readable name (Geonames stores a 2-letter continent code).
pub fn country_context(iso2s: &[&str]) -> CliResult<Vec<Option<CountryContext>>> {
    with_geocode_engine(|engine| {
        iso2s
            .iter()
            .map(|&iso2| {
                let countryrecord = engine.country_info(iso2)?;
                let continent = continent_name(&countryrecord.info.continent).to_string();
                if continent.is_empty() {
                    return None;
                }
                Some(CountryContext { continent })
            })
            .collect()
    })
}

/// Expand a Geonames 2-letter continent code to its human-readable name; unknown codes pass through
/// unchanged (so a future/unrecognized code still shows something rather than nothing).
fn continent_name(code: &str) -> &str {
    match code {
        "AF" => "Africa",
        "AN" => "Antarctica",
        "AS" => "Asia",
        "EU" => "Europe",
        "NA" => "North America",
        "OC" => "Oceania",
        "SA" => "South America",
        other => other,
    }
}

#[inline]
fn lookup_us_state_fips_code(state: &str) -> Option<&'static str> {
    US_STATES_FIPS_CODES.get(state).copied()
}

/// Derive `(us_state_fips, us_county_fips)` string codes for a matched city, for the enriched
/// [`GeoLabel`]. Thin wrapper over [`us_fips_from_codes`] that pulls the admin1/admin2 codes off
/// the record.
fn us_fips_strings(cityrecord: &CitiesRecord) -> (String, String) {
    us_fips_from_codes(
        cityrecord
            .admin_division
            .as_ref()
            .map(|admin1| &*admin1.code),
        cityrecord
            .admin2_division
            .as_ref()
            .map(|admin2| &*admin2.code),
    )
}

/// Pure FIPS derivation from Geonames admin1/admin2 codes (e.g. `US.PA` / `US.PA.003`).
/// `us_state_fips` is the 2-digit state FIPS ("42") or `""` when the admin1 isn't a US state;
/// `us_county_fips` is the 5-digit combined state+county FIPS ("42003") or `""` when no valid US
/// county code is present. The county segment is the LAST dotted component of the admin2 code (e.g.
/// `US.NY.119` -> `119`), which is robust regardless of its leading digit.
fn us_fips_from_codes(admin1_code: Option<&str>, admin2_code: Option<&str>) -> (String, String) {
    let state_fips = admin1_code
        .and_then(|code| code.strip_prefix("US."))
        .and_then(lookup_us_state_fips_code)
        .unwrap_or_default();
    if state_fips.is_empty() {
        // without a state FIPS a combined county FIPS can't be formed
        return (String::new(), String::new());
    }
    let us_county_fips = admin2_code
        .filter(|code| code.starts_with("US."))
        .and_then(|code| code.rsplit('.').next())
        .filter(|county| !county.is_empty() && county.bytes().all(|b| b.is_ascii_digit()))
        .map_or_else(String::new, |county| format!("{state_fips}{county:0>3}"));
    (state_fips.to_string(), us_county_fips)
}

fn get_us_fips_codes(cityrecord: &CitiesRecord, nameslang: &NamesLang) -> serde_json::Value {
    // None when admin1 is not a US state code; the 2-letter state code otherwise
    let us_state_code: Option<&str> = cityrecord
        .admin_division
        .as_ref()
        .and_then(|admin1| admin1.code.strip_prefix("US."));

    // emit JSON null when the admin1 isn't a US state or the code is unrecognized
    let us_state_fips_code = us_state_code
        .and_then(lookup_us_state_fips_code)
        .map_or(serde_json::Value::Null, |code| {
            serde_json::Value::String(code.to_string())
        });

    // emit JSON null when admin2 isn't a valid US county code (US.XX.NNN format).
    // require ASCII so the byte-index slice at 7 lands on a char boundary
    let us_county_fips_code = match cityrecord.admin2_division.as_ref() {
        Some(admin2)
            if admin2.code.is_ascii()
                && admin2.code.len() == 9
                && admin2.code.starts_with("US.") =>
        {
            // skip the "US." prefix and 2-letter state code (7 ASCII bytes),
            // e.g. US.NY.061 -> 061
            serde_json::Value::String(format!("{:0>3}", &admin2.code[7..]))
        },
        _ => serde_json::Value::Null,
    };
    json!(
    {
        "us_state_code": us_state_code,
        "us_state_name": nameslang.admin1name,
        "us_state_fips_code": us_state_fips_code,
        "us_county": nameslang.admin2name,
        "us_county_fips_code": us_county_fips_code,
    })
}

#[inline]
fn add_fields(record: &mut csv::StringRecord, value: &str, count: u8) {
    (0..count).for_each(|_| {
        record.push_field(value);
    });
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn parse_relative_age_units() {
        assert_eq!(parse_relative_age("30s"), Some(Duration::from_secs(30)));
        assert_eq!(parse_relative_age("5m"), Some(Duration::from_secs(300)));
        assert_eq!(parse_relative_age("2h"), Some(Duration::from_secs(7_200)));
        assert_eq!(parse_relative_age("3d"), Some(Duration::from_secs(259_200)));
        assert_eq!(parse_relative_age("1w"), Some(Duration::from_secs(604_800)));
    }

    #[test]
    fn us_fips_from_codes_derivation() {
        // state + county -> 5-digit combined FIPS (Allegheny County, PA)
        assert_eq!(
            us_fips_from_codes(Some("US.PA"), Some("US.PA.003")),
            ("42".to_string(), "42003".to_string())
        );
        // county code whose LAST segment does NOT lead with 0 must not be mangled (New York County)
        assert_eq!(
            us_fips_from_codes(Some("US.NY"), Some("US.NY.061")),
            ("36".to_string(), "36061".to_string())
        );
        // a 2-digit county segment is left-padded to 3 before combining
        assert_eq!(
            us_fips_from_codes(Some("US.CA"), Some("US.CA.37")),
            ("06".to_string(), "06037".to_string())
        );
        // state only (no admin2) -> state FIPS, empty county FIPS
        assert_eq!(
            us_fips_from_codes(Some("US.PA"), None),
            ("42".to_string(), String::new())
        );
        // non-US admin1 -> both empty (no state FIPS, so no county FIPS either)
        assert_eq!(
            us_fips_from_codes(Some("CA.ON"), Some("CA.ON.something")),
            (String::new(), String::new())
        );
        // a non-US admin2 alongside a US state is ignored
        assert_eq!(
            us_fips_from_codes(Some("US.PA"), Some("XX.PA.003")),
            ("42".to_string(), String::new())
        );
    }

    #[test]
    fn continent_name_expands_known_codes() {
        assert_eq!(continent_name("NA"), "North America");
        assert_eq!(continent_name("EU"), "Europe");
        assert_eq!(continent_name("SA"), "South America");
        // unknown/empty codes pass through unchanged
        assert_eq!(continent_name("ZZ"), "ZZ");
        assert_eq!(continent_name(""), "");
    }

    #[test]
    fn parse_relative_age_case_and_whitespace() {
        // unit is case-insensitive and whitespace around/within is tolerated
        assert_eq!(
            parse_relative_age("30D"),
            Some(Duration::from_secs(2_592_000))
        );
        assert_eq!(
            parse_relative_age("  30 d  "),
            Some(Duration::from_secs(2_592_000))
        );
    }

    #[test]
    fn parse_relative_age_non_matches() {
        // not a relative-age form -> None (caller falls back to date parsing)
        assert_eq!(parse_relative_age("2025-01-01"), None);
        assert_eq!(parse_relative_age("garbage"), None);
        assert_eq!(parse_relative_age("30"), None); // missing unit
        assert_eq!(parse_relative_age("30y"), None); // unsupported unit
    }

    #[test]
    fn parse_relative_age_overflow_is_none() {
        // a relative-age form whose seconds overflow u64 -> None
        assert_eq!(parse_relative_age("99999999999999999999999999w"), None);
    }

    #[test]
    fn resolve_older_than_relative_age() {
        assert_eq!(
            resolve_older_than("30d").unwrap(),
            Duration::from_secs(2_592_000)
        );
    }

    #[test]
    fn resolve_older_than_absolute_date() {
        // a date safely in the past resolves to a large positive age
        let age = resolve_older_than("2000-01-01").unwrap();
        assert!(age.as_secs() > 600_000_000);
    }

    #[test]
    fn resolve_older_than_future_date_rejected() {
        let err = resolve_older_than("2099-01-01").unwrap_err();
        assert!(matches!(&err, CliError::IncorrectUsage(m) if m.contains("in the future")));
    }

    #[test]
    fn resolve_older_than_pre_epoch_rejected() {
        let err = resolve_older_than("1850-01-01").unwrap_err();
        assert!(matches!(&err, CliError::IncorrectUsage(m) if m.contains("Unix epoch")));
    }

    #[test]
    fn resolve_older_than_overflow_rejected() {
        // a relative-age form that overflows reports the overflow, not a date-parse error
        let err = resolve_older_than("99999999999999999999999999w").unwrap_err();
        assert!(matches!(&err, CliError::IncorrectUsage(m) if m.contains("too large")));
    }

    #[test]
    fn resolve_older_than_garbage_rejected() {
        let err = resolve_older_than("not-a-date").unwrap_err();
        assert!(matches!(&err, CliError::IncorrectUsage(m) if m.contains("not a relative age")));
    }
}
