Gitingest

Include files under: 50kb

Summary

Repository: fabianlindfors/reshape
Files analyzed: 47

Estimated tokens: 55.5k

Download

Directory Structure

Directory structure:

└── fabianlindfors-reshape/

    ├── README.md

    ├── Cargo.toml

    ├── Dockerfile

    ├── LICENSE.md

    ├── .dockerignore

    ├── src/

    │   ├── db.rs

    │   ├── helpers.rs

    │   ├── lib.rs

    │   ├── main.rs

    │   ├── schema.rs

    │   ├── state.rs

    │   └── migrations/

    │       ├── add_column.rs

    │       ├── add_foreign_key.rs

    │       ├── add_index.rs

    │       ├── alter_column.rs

    │       ├── common.rs

    │       ├── create_enum.rs

    │       ├── create_table.rs

    │       ├── custom.rs

    │       ├── mod.rs

    │       ├── remove_column.rs

    │       ├── remove_enum.rs

    │       ├── remove_foreign_key.rs

    │       ├── remove_index.rs

    │       ├── remove_table.rs

    │       └── rename_table.rs

    ├── tests/

    │   ├── add_column.rs

    │   ├── add_foreign_key.rs

    │   ├── add_index.rs

    │   ├── alter_column.rs

    │   ├── common.rs

    │   ├── complex.rs

    │   ├── create_enum.rs

    │   ├── create_table.rs

    │   ├── custom.rs

    │   ├── failure.rs

    │   ├── remove_column.rs

    │   ├── remove_enum.rs

    │   ├── remove_foreign_key.rs

    │   ├── remove_index.rs

    │   ├── remove_table.rs

    │   └── rename_table.rs

    └── .github/

        └── workflows/

            ├── lint.yaml

            ├── publish-crate.yaml

            ├── publish-docker.yaml

            ├── release.yaml

            └── test.yaml

Files Content

================================================
FILE: README.md
================================================
# Reshape

[![Test status badge](https://github.com/fabianlindfors/Reshape/actions/workflows/test.yaml/badge.svg)](https://github.com/fabianlindfors/reshape/actions/workflows/test.yaml) [![Latest release](https://shields.io/github/v/release/fabianlindfors/reshape?display_name=tag&sort=semver&color=blue)](https://github.com/fabianlindfors/reshape/releases)

> If you like Postgres, you might also be interested in [pgfdb](https://github.com/fabianlindfors/pgfdb). pgfdb is an experimental project that turns Postgres into a distributed and horizontally scalable database.

Reshape is an easy-to-use, zero-downtime schema migration tool for Postgres. It automatically handles complex migrations that would normally require downtime or manual multi-step changes. During a migration, Reshape ensures both the old and new schema are available at the same time, allowing you to gradually roll out your application. It will also perform all changes without excessive locking, avoiding downtime caused by blocking other queries. For a more thorough introduction to Reshape, check out the [introductory blog post](https://fabianlindfors.se/blog/schema-migrations-in-postgres-using-reshape/).

Designed for Postgres 12 and later.

- [How it works](#how-it-works)
- [Getting started](#getting-started)
  - [Installation](#installation)
  - [Creating your first migration](#creating-your-first-migration)
  - [Preparing your application](#preparing-your-application)
  - [Running your migration](#running-your-migration)
  - [Using during development](#using-during-development)
- [Writing migrations](#writing-migrations)
  - [Basics](#basics)
  - [Tables](#tables)
    - [Create table](#create-table)
    - [Rename table](#rename-table)
    - [Remove table](#remove-table)
    - [Add foreign key](#add-foreign-key)
    - [Remove foreign key](#remove-foreign-key)
  - [Columns](#columns)
    - [Add column](#add-column)
    - [Alter column](#alter-column)
    - [Remove column](#remove-column)
  - [Indices](#indices)
    - [Add index](#add-index)
    - [Remove index](#remove-index)
  - [Enums](#enums)
    - [Create enum](#create-enum)
    - [Remove enum](#remove-enum)
  - [Custom](#custom)
  - [Complex changes across tables](#complex-changes-across-tables)
- [Commands and options](#commands-and-options)
  - [`reshape migration start`](#reshape-migration-start)
  - [`reshape migration complete`](#reshape-migration-complete)
  - [`reshape migration abort`](#reshape-migration-abort)
  - [`reshape schema-query`](#reshape-schema-query)
  - [Connection options](#connection-options)
- [License](#license)

## How it works

Reshape works by creating views that encapsulate the underlying tables, which your application will interact with. During a migration, Reshape will automatically create a new set of views and set up triggers to translate inserts and updates between the old and new schema. This means that every deployment is a three-phase process:

1. **Start migration** (`reshape migration start`): Sets up views and triggers to ensure both the new and old schema are usable at the same time.
2. **Roll out application**: Your application can be gradually rolled out without downtime. The existing deployment will continue using the old schema whilst the new deployment uses the new schema.
3. **Complete migration** (`reshape migration complete`): Removes the old schema and any intermediate data and triggers.

If the application deployment fails, you should run `reshape migration abort` which will roll back any changes made by `reshape migration start` without losing data.

## Getting started

### Installation

#### Binaries

Binaries are available for macOS and Linux under [Releases](https://github.com/fabianlindfors/reshape/releases).

#### Cargo

Reshape can be installed using [Cargo](https://doc.rust-lang.org/cargo/) (requires Rust 1.58 or later):

```shell
cargo install reshape
```

#### Docker

Reshape is available as a Docker image on [Docker Hub](https://hub.docker.com/repository/docker/fabianlindfors/reshape).

```shell
docker run -v $(pwd):/usr/share/app fabianlindfors/reshape reshape migration start
```

### Creating your first migration

Each migration should be stored as a separate file in a `migrations/` directory. The files can be in either JSON or TOML format and the name of the file will become the name of your migration. We recommend prefixing every migration with an incrementing number as migrations are sorted by file name.

Let's create a simple migration to set up a new table `users` with two fields, `id` and `name`. We'll create a file called `migrations/1_create_users_table.toml`:

```toml
[[actions]]
type = "create_table"
name = "users"
primary_key = ["id"]

	[[actions.columns]]
	name = "id"
	type = "INTEGER"
	generated = "ALWAYS AS IDENTITY"

	[[actions.columns]]
	name = "name"
	type = "TEXT"
```

This is the equivalent of running `CREATE TABLE users (id INTEGER GENERATED ALWAYS AS IDENTITY, name TEXT)`.

### Preparing your application

Reshape relies on your application using a specific schema. When establishing the connection to Postgres in your application, you need to run a query to select the most recent schema. The simplest way to do this is to use one of the helper libraries:

- [Rust](https://github.com/fabianlindfors/reshape-helper)
- [Ruby (and Rails)](https://github.com/fabianlindfors/reshape-ruby)
- [Python (and Django)](https://github.com/fabianlindfors/reshape-python)
- [Go](https://github.com/leourbina/reshape-helper)

If your application is not using one of the languages with an available helper library, you can instead generate the query with the command: `reshape schema-query`. To pass it along to your application, you can for example use an environment variable in your run script: `RESHAPE_SCHEMA_QUERY=$(reshape schema-query)`. Then in your application:

```python
# Example for Python
reshape_schema_query = os.getenv("RESHAPE_SCHEMA_QUERY")
db.execute(reshape_schema_query)
```

### Running your migration

To create your new `users` table, run:

```bash
reshape migration start --complete
```

We use the `--complete` flag to automatically complete the migration. During a production deployment, you should first run `reshape migration start` followed by `reshape migration complete` once your application has been fully rolled out.

If nothing else is specified, Reshape will try to connect to a Postgres database running on `localhost` using `postgres` as both username and password. See [Connection options](#connection-options) for details on how to change the connection settings.

### Using during development

When adding new migrations during development, we recommend running `reshape migration start` but skipping `reshape migration complete`. This way, the new migrations can be iterated on by updating the migration file and running `reshape migration abort` followed by `reshape migration start`.

## Writing migrations

### Basics

Every migration consists of one or more actions. The actions will be run sequentially. Here's an example of a migration with two actions to create two tables, `customers` and `products`:

```toml
[[actions]]
type = "create_table"
name = "customers"
primary_key = ["id"]

	[[actions.columns]]
	name = "id"
	type = "INTEGER"
	generated = "ALWAYS AS IDENTITY"

[[actions]]
type = "create_table"
name = "products"
primary_key = ["sku"]

	[[actions.columns]]
	name = "sku"
	type = "TEXT"
```

Every action has a `type`. The supported types are detailed below.

### Tables

#### Create table

The `create_table` action will create a new table with the specified columns, indices and constraints. You can optionally provide an `up` option to backfill values from an existing table.

_Example: create a `customers` table with a few columns and a primary key_

```toml
[[actions]]
type = "create_table"
name = "customers"
primary_key = ["id"]

	[[actions.columns]]
	name = "id"
	type = "INTEGER"
	generated = "ALWAYS AS IDENTITY"

	[[actions.columns]]
	name = "name"
	type = "TEXT"

	# Columns default to nullable
	nullable = false

	# default can be any valid SQL value, in this case a string literal
	default = "'PLACEHOLDER'"
```

_Example: create `users` and `items` tables with a foreign key between them_

```toml
[[actions]]
type = "create_table"
name = "users"
primary_key = ["id"]

	[[actions.columns]]
	name = "id"
	type = "INTEGER"
	generated = "ALWAYS AS IDENTITY"

[[actions]]
type = "create_table"
name = "items"
primary_key = ["id"]

	[[actions.columns]]
	name = "id"
	type = "INTEGER"
	generated = "ALWAYS AS IDENTITY"

	[[actions.columns]]
	name = "user_id"
	type = "INTEGER"

	[[actions.foreign_keys]]
	columns = ["user_id"]
	referenced_table = "users"
	referenced_columns = ["id"]
```

_Example: create `profiles` table based on existing `users` table_

```toml
[[actions]]
type = "create_table"
name = "profiles"
primary_key = ["user_id"]

	[[actions.columns]]
	name = "user_id"
	type = "INTEGER"

	[[actions.columns]]
	name = "user_email"
	type = "TEXT"

	# Backfill from `users` table and copy `users.email` to `user_email` column
	# This will perform an upsert based on the primary key to avoid duplicate rows
	[actions.up]
	table = "users"
	values = { user_id = "id", user_email = "email" }
```

#### Rename table

The `rename_table` action will change the name of an existing table.

_Example: change name of `users` table to `customers`_

```toml
[[actions]]
type = "rename_table"
table = "users"
new_name = "customers"
```

#### Remove table

The `remove_table` action will remove an existing table.

_Example: remove `users` table_

```toml
[[actions]]
type = "remove_table"
table = "users"
```

#### Add foreign key

The `add_foreign_key` action will add a foreign key between two existing tables. The migration will fail if the existing column values aren't valid references.

_Example: create foreign key from `items` to `users` table_

```toml
[[actions]]
type = "add_foreign_key"
table = "items"

	[actions.foreign_key]
	columns = ["user_id"]
	referenced_table = "users"
	referenced_columns = ["id"]
```

#### Remove foreign key

The `remove_foreign_key` action will remove an existing foreign key. The foreign key will only be removed once the migration is completed, which means that your new application must continue to adhere to the foreign key constraint.

_Example: remove foreign key `items_user_id_fkey` from `users` table_

```toml
[[actions]]
type = "remove_foreign_key"
table = "items"
foreign_key = "items_user_id_fkey"
```

### Columns

#### Add column

The `add_column` action will add a new column to an existing table. You can optionally provide an `up` setting. This should be an SQL expression which will be run for all existing rows to backfill the new column. `up` may also reference another table to perform cross-table migrations (see ["Complex changes across tables"](#complex-changes-across-tables)).

_Example: add a new column `reference` to table `products`_

```toml
[[actions]]
type = "add_column"
table = "products"

	[actions.column]
	name = "reference"
	type = "INTEGER"
	nullable = false
	default = "10"
```

_Example: replace an existing `name` column with two new columns, `first_name` and `last_name`_

```toml
[[actions]]
type = "add_column"
table = "users"

# Extract the first name from the existing name column
up = "(STRING_TO_ARRAY(name, ' '))[1]"

	[actions.column]
	name = "first_name"
	type = "TEXT"


[[actions]]
type = "add_column"
table = "users"

# Extract the last name from the existing name column
up = "(STRING_TO_ARRAY(name, ' '))[2]"

	[actions.column]
	name = "last_name"
	type = "TEXT"


[[actions]]
type = "remove_column"
table = "users"
column = "name"

# Reconstruct name column by concatenating first and last name
down = "first_name || ' ' || last_name"
```

_Example: extract nested value from unstructured JSON `data` column to new `name` column_

```toml
[[actions]]
type = "add_column"
table = "users"

# #>> '{}' converts the JSON string value to TEXT
up = "data['path']['to']['value'] #>> '{}'"

	[actions.column]
	name = "name"
	type = "TEXT"
```

_Example: duplicate `email` column from `users` to `profiles` table_

```toml
# `profiles` has `user_id` column which maps to `users.id`
[[actions]]
type = "add_column"
table = "profiles"

	[actions.column]
	name = "email"
	type = "TEXT"
	nullable = false

	# When `users` is updated in the old schema, we write the email value to `profiles`
	[actions.up]
	table = "users"
	value = "email"
	where = "user_id = id"
```

#### Alter column

The `alter_column` action enables many different changes to an existing column, for example renaming, changing type and changing existing values.

When performing more complex changes than a rename, `up` and `down` should be provided. These should be SQL expressions which determine how to transform between the new and old version of the column. Inside those expressions, you can reference the current column value by the column name.

_Example: rename `last_name` column on `users` table to `family_name`_

```toml
[[actions]]
type = "alter_column"
table = "users"
column = "last_name"

	[actions.changes]
	name = "family_name"
```

_Example: change the type of `reference` column from `INTEGER` to `TEXT`_

```toml
[[actions]]
type = "alter_column"
table = "users"
column = "reference"

up = "CAST(reference AS TEXT)" # Converts from integer value to text
down = "CAST(reference AS INTEGER)" # Converts from text value to integer

	[actions.changes]
	type = "TEXT" # Previous type was 'INTEGER'
```

_Example: increment all values of an `index` column by one_

```toml
[[actions]]
type = "alter_column"
table = "users"
column = "index"

up = "index + 1" # Increment for new schema
down = "index - 1" # Decrement to revert for old schema

	[actions.changes]
	name = "index"
```

_Example: make `name` column not nullable_

```toml
[[actions]]
type = "alter_column"
table = "users"
column = "name"

# Use "N/A" for any rows that currently have a NULL name
up = "COALESCE(name, 'N/A')"

	[actions.changes]
	nullable = false
```

_Example: change default value of `created_at` column to current time_

```toml
[[actions]]
type = "alter_column"
table = "users"
column = "created_at"

	[actions.changes]
	default = "NOW()"
```

#### Remove column

The `remove_column` action will remove an existing column from a table. You can optionally provide a `down` setting. This should be an SQL expression which will be used to determine values for the old schema when inserting or updating rows using the new schema. `down` may also reference another table to perform cross-table migrations (see ["Complex changes across tables"](#complex-changes-across-tables)) . The `down` setting must be provided when the removed column is `NOT NULL` or doesn't have a default value.

Any indices that cover the column will be removed.

_Example: remove column `name` from table `users`_

```toml
[[actions]]
type = "remove_column"
table = "users"
column = "name"

# Use a default value of "N/A" for the old schema when inserting/updating rows
down = "'N/A'"
```

_Example: remove `email` column from `users` table and use column from `profiles` table instead_

```toml
[[actions]]
type = "remove_column"
table = "users"
column = "email"

	# Our application will use the `profiles.email` column instead
	# For backwards compatibility, we will write back to the removed `email` column whenever `profiles` is changed
	[actions.down]
	table = "profiles"
	value = "profiles.email"
	where = "users.id = profiles.user_id"
```

### Indices

#### Add index

The `add_index` action will add a new index to an existing table.

_Example: create a `users` table with a unique index on the `name` column_

```toml
[[actions]]
type = "create_table"
name = "users"
primary_key = "id"

	[[actions.columns]]
	name = "id"
	type = "INTEGER"
	generated = "ALWAYS AS IDENTITY"

	[[actions.columns]]
	name = "name"
	type = "TEXT"

[[actions]]
type = "add_index"
table = "users"

	[actions.index]
	name = "name_idx"
	columns = ["name"]

	# Defaults to false
	unique = true
```

_Example: add GIN index to `data` column on `products` table_

```toml
[[actions]]
type = "add_index"
table = "products"

	[actions.index]
	name = "data_idx"
	columns = ["data"]

	# One of: btree (default), hash, gist, spgist, gin, brin
	type = "gin"
```

#### Remove index

The `remove_index` action will remove an existing index. The index won't actually be removed until the migration is completed.

_Example: remove the `name_idx` index_

```toml
[[actions]]
type = "remove_index"
index = "name_idx"
```

### Enums

#### Create enum

The `create_enum` action will create a new [enum type](https://www.postgresql.org/docs/current/datatype-enum.html) with the specified values.

_Example: add a new `mood` enum type with three possible values_

```toml
[[actions]]
type = "create_enum"
name = "mood"
values = ["happy", "ok", "sad"]
```

#### Remove enum

The `remove_enum` action will remove an existing [enum type](https://www.postgresql.org/docs/current/datatype-enum.html). Make sure all usages of the enum has been removed before running the migration. The enum will only be removed once the migration is completed.

_Example: remove the `mood` enum type_

```toml
[[actions]]
type = "remove_enum"
enum = "mood"
```

### Custom

The `custom` action lets you create a migration which runs custom SQL. It should be used with great care as it provides no guarantees of zero-downtime and will simply run whatever SQL is provided. Use other actions whenever possible as they are explicitly designed for zero downtime.

There are three optional settings available which all accept SQL queries. All queries need to be idempotent, for example by using `IF NOT EXISTS` wherever available.

- `start`: run when a migration is started using `reshape migration start`
- `complete`: run when a migration is completed using `reshape migration complete`
- `abort`: run when a migration is aborted using `reshape migration abort`

_Example: enable PostGIS and pg_stat_statements extensions_

```toml
[[actions]]
type = "custom"

start = """
	CREATE EXTENSION IF NOT EXISTS postgis;
	CREATE EXTENSION IF NOT EXISTS pg_stat_statements;
"""

abort = """
	DROP EXTENSION IF EXISTS postgis;
	DROP EXTENSION IF EXISTS pg_stat_statements;
"""
```

### Complex changes across tables

The `up` and `down` options available when creating tables, adding columns and removing columns can also perform more complex changes that span tables.

_Example: move `email` column from `users` to `profiles` table_

```toml
[[actions]]
type = "add_column"
table = "profiles"

	[actions.column]
	name = "email"
	type = "TEXT"
	nullable = false

	# When `users` is updated in the old schema, we write the email value to `profiles`
	[actions.up]
	table = "users"
	value = "users.email"
	where = "profiles.user_id = users.id"

[[actions]]
type = "remove_column"
table = "users"
column = "email"

	# When `profiles` is changed in the new schema, we write the email address back to the removed column
	[actions.down]
	table = "profiles"
	value = "profiles.email"
	where = "users.id = profiles.user_id"
```

_Example: turn a 1:N relationship between `users` and `accounts` into N:M and change the format of the associated `role`_

```toml
# Add `user_account_connections` as a junction table
[[actions]]
type = "create_table"
name = "user_account_connections"
primary_key = ["account_id", "user_id"]

	[[actions.columns]]
	name = "account_id"
	type = "INTEGER"

	[[actions.columns]]
	name = "user_id"
	type = "INTEGER"

	# `role` is currently stored directly on the `users` table but is part of the relationship
	[[actions.columns]]
	name = "role"
	type = "TEXT"
	nullable = false

	# Backfill the new table from `users` and uppercase the `role`
	[actions.up]
	table = "users"
	values = { user_id = "id", account_id = "account_id", role = "UPPER(account_role)" }
	where = "user_account_connections.user_id = users.id"

[[actions]]
type = "remove_column"
table = "users"
column = "account_id"

	# When `user_account_connections` is updated, we write the `account_id` back to `users`
	[actions.down]
	table = "user_account_connections"
	value = "user_account_connections.account_id"
	where = "users.id = user_account_connections.user_id"

[[actions]]
type = "remove_column"
table = "users"
column = "account_role"

	# When `user_account_connections` is updated, we write the lowercase role back to `users`
	[actions.down]
	table = "user_account_connections"
	value = "LOWER(user_account_connections.role)"
	where = "users.id = user_account_connections.user_id"
```

## Commands and options

### `reshape migration start`

Starts a new migration, applying all migrations under `migrations/` that haven't yet been applied. After the command has completed, both the old and new schema will be usable at the same time. When you have rolled out the new version of your application which uses the new schema, you should run `reshape migration complete`.

#### Options

_See also [Connection options](#connection-options)_

| Option             | Default       | Description                                                                                                     |
| ------------------ | ------------- | --------------------------------------------------------------------------------------------------------------- |
| `--complete`, `-c` | `false`       | Automatically complete migration after applying it.                                                             |
| `--dirs`           | `migrations/` | Directories to search for migration files. Multiple directories can be specified using `--dirs dir1 dir2 dir3`. |

### `reshape migration complete`

Completes migrations previously started with `reshape migration complete`.

#### Options

See [Connection options](#connection-options)

### `reshape migration abort`

Aborts any migrations which haven't yet been completed.

#### Options

See [Connection options](#connection-options)

### `reshape schema-query`

Generates the SQL query you need to run in your application before using the database. This command does not require a database connection. Instead it will generate the query based on the latest migration in the `migrations/` directory (or the directories specified by `--dirs`).

If your application is written in Rust, we recommend using the [Rust helper library](https://github.com/fabianlindfors/reshape-helper/) instead.

The query should look something like `SET search_path TO migration_1_initial_migration`.

#### Options

| Option   | Default       | Description                                                                                                     |
| -------- | ------------- | --------------------------------------------------------------------------------------------------------------- |
| `--dirs` | `migrations/` | Directories to search for migration files. Multiple directories can be specified using `--dirs dir1 dir2 dir3`. |

### Connection options

The options below can be used with all commands that communicate with Postgres. Use either a [connection URL](https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNSTRING) or specify each connection option individually.

All options can also be set using environment variables instead of flags. If a `.env` file exists, then variables will be automatically loaded from there.

| Option       | Default     | Environment variable | Description                                 |
| ------------ | ----------- | -------------------- | ------------------------------------------- |
| `--url`      |             | `DB_URL`             | URL to your Postgres database               |
| `--host`     | `localhost` | `DB_HOST`            | Hostname to use when connecting to Postgres |
| `--port`     | `5432`      | `DB_PORT`            | Port which Postgres is listening on         |
| `--database` | `postgres`  | `DB_NAME`            | Database name                               |
| `--username` | `postgres`  | `DB_USERNAME`        | Postgres username                           |
| `--password` | `postgres`  | `DB_PASSWORD`        | Postgres password                           |

## License

Reshape is released under the [MIT license](https://choosealicense.com/licenses/mit/).



================================================
FILE: Cargo.toml
================================================
[package]
name = "reshape"
version = "0.7.0"
description = "An easy-to-use, zero-downtime schema migration tool for Postgres"
homepage = "https://github.com/fabianlindfors/reshape"
documentation = "https://github.com/fabianlindfors/reshape"
repository = "https://github.com/fabianlindfors/reshape"
license = "MIT"
keywords = ["postgres", "migrations"]
edition = "2021"
authors = ["Fabian Lindfors"]
rust-version = "1.70"

[dependencies]
postgres = { version = "0.19.2", features = ["with-serde_json-1"] }
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
typetag = "0.1.7"
anyhow = { version = "1.0.44", features = ["backtrace"] }
clap = { version = "3.1.9", features = ["derive"] }
toml = "0.5"
version = "3.0.0"
colored = "2"
rand = "0.8"
dotenv = "0.15.0"
lexical-sort = "0.3.1"



================================================
FILE: Dockerfile
================================================
FROM rust:1.70.0 AS builder
WORKDIR /usr/src/reshape
COPY . .
RUN cargo build --release

FROM debian:bullseye AS runtime
WORKDIR /usr/share/app
COPY --from=builder /usr/src/reshape/target/release/reshape /usr/local/bin/reshape
CMD ["reshape"]


================================================
FILE: LICENSE.md
================================================
MIT License

Copyright (c) 2022 Fabian Lindfors

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.



================================================
FILE: .dockerignore
================================================
/target
Cargo.lock


================================================
FILE: src/db.rs
================================================
use std::{cmp::min, time::Duration};

use anyhow::{anyhow, Context};
use postgres::{types::ToSql, NoTls, Row};
use rand::prelude::*;

// DbLocker wraps a regular DbConn, only allowing access using the
// `lock` method. This method will acquire the advisory lock before
// allowing access to the database, and then release it afterwards.
//
// We use advisory locks to avoid multiple Reshape instances working
// on the same database as the same time. DbLocker is the only way to
// get a DbConn which ensures that all database access is protected by
// a lock.
//
// Postgres docs on advisory locks:
//   https://www.postgresql.org/docs/current/explicit-locking.html#ADVISORY-LOCKS
pub struct DbLocker {
    client: DbConn,
}

impl DbLocker {
    // Advisory lock keys in Postgres are 64-bit integers.
    // The key we use was chosen randomly.
    const LOCK_KEY: i64 = 4036779288569897133;

    pub fn connect(config: &postgres::Config) -> anyhow::Result<Self> {
        let mut pg = config.connect(NoTls)?;

        // When running DDL queries that acquire locks, we risk causing a "lock queue".
        // When attempting to acquire a lock, Postgres will wait for any long running queries to complete.
        // At the same time, it will block other queries until the lock has been acquired and released.
        // This has the bad effect of the long-running query blocking other queries because of us, forming
        // a queue of other queries until we release our lock.
        //
        // We set the lock_timeout setting to avoid this. This puts an upper bound for how long Postgres will
        // wait to acquire locks and also the maximum amount of time a long-running query can block other queries.
        // We should also add automatic retries to handle these timeouts gracefully.
        //
        // Reference: https://medium.com/paypal-tech/postgresql-at-scale-database-schema-changes-without-downtime-20d3749ed680
        //
        // TODO: Make lock_timeout configurable
        pg.simple_query("SET lock_timeout = '1s'")
            .context("failed to set lock_timeout")?;

        Ok(Self {
            client: DbConn::new(pg),
        })
    }

    pub fn lock(
        &mut self,
        f: impl FnOnce(&mut DbConn) -> anyhow::Result<()>,
    ) -> anyhow::Result<()> {
        self.acquire_lock()?;
        let result = f(&mut self.client);
        self.release_lock()?;

        result
    }

    fn acquire_lock(&mut self) -> anyhow::Result<()> {
        let success = self
            .client
            .query(&format!("SELECT pg_try_advisory_lock({})", Self::LOCK_KEY))?
            .first()
            .ok_or_else(|| anyhow!("unexpectedly failed when acquiring advisory lock"))
            .map(|row| row.get::<'_, _, bool>(0))?;

        if success {
            Ok(())
        } else {
            Err(anyhow!("another instance of Reshape is already running"))
        }
    }

    fn release_lock(&mut self) -> anyhow::Result<()> {
        self.client
            .query(&format!("SELECT pg_advisory_unlock({})", Self::LOCK_KEY))?
            .first()
            .ok_or_else(|| anyhow!("unexpectedly failed when releasing advisory lock"))?;
        Ok(())
    }
}

pub trait Conn {
    fn run(&mut self, query: &str) -> anyhow::Result<()>;
    fn query(&mut self, query: &str) -> anyhow::Result<Vec<Row>>;
    fn query_with_params(
        &mut self,
        query: &str,
        params: &[&(dyn ToSql + Sync)],
    ) -> anyhow::Result<Vec<Row>>;
    fn transaction(&mut self) -> anyhow::Result<Transaction>;
}

pub struct DbConn {
    client: postgres::Client,
}

impl DbConn {
    fn new(client: postgres::Client) -> Self {
        DbConn { client }
    }
}

impl Conn for DbConn {
    fn run(&mut self, query: &str) -> anyhow::Result<()> {
        retry_automatically(|| self.client.batch_execute(query))?;
        Ok(())
    }

    fn query(&mut self, query: &str) -> anyhow::Result<Vec<Row>> {
        let rows = retry_automatically(|| self.client.query(query, &[]))?;
        Ok(rows)
    }

    fn query_with_params(
        &mut self,
        query: &str,
        params: &[&(dyn ToSql + Sync)],
    ) -> anyhow::Result<Vec<Row>> {
        let rows = retry_automatically(|| self.client.query(query, params))?;
        Ok(rows)
    }

    fn transaction(&mut self) -> anyhow::Result<Transaction> {
        let transaction = self.client.transaction()?;
        Ok(Transaction { transaction })
    }
}

pub struct Transaction<'a> {
    transaction: postgres::Transaction<'a>,
}

impl Transaction<'_> {
    pub fn commit(self) -> anyhow::Result<()> {
        self.transaction.commit()?;
        Ok(())
    }

    pub fn rollback(self) -> anyhow::Result<()> {
        self.transaction.rollback()?;
        Ok(())
    }
}

impl Conn for Transaction<'_> {
    fn run(&mut self, query: &str) -> anyhow::Result<()> {
        self.transaction.batch_execute(query)?;
        Ok(())
    }

    fn query(&mut self, query: &str) -> anyhow::Result<Vec<Row>> {
        let rows = self.transaction.query(query, &[])?;
        Ok(rows)
    }

    fn query_with_params(
        &mut self,
        query: &str,
        params: &[&(dyn ToSql + Sync)],
    ) -> anyhow::Result<Vec<Row>> {
        let rows = self.transaction.query(query, params)?;
        Ok(rows)
    }

    fn transaction(&mut self) -> anyhow::Result<Transaction> {
        let transaction = self.transaction.transaction()?;
        Ok(Transaction { transaction })
    }
}

// Retry a database operation with exponential backoff and jitter
fn retry_automatically<T>(
    mut f: impl FnMut() -> Result<T, postgres::Error>,
) -> Result<T, postgres::Error> {
    const STARTING_WAIT_TIME: u64 = 100;
    const MAX_WAIT_TIME: u64 = 3_200;
    const MAX_ATTEMPTS: u32 = 10;

    let mut rng = rand::thread_rng();
    let mut attempts = 0;
    loop {
        let result = f();

        let error = match result {
            Ok(_) => return result,
            Err(err) => err,
        };

        // If we got a database error, we check if it's retryable.
        // If we didn't get a database error, then it's most likely some kind of connection
        // error which should also be retried.
        if let Some(db_error) = error.as_db_error() {
            if !error_retryable(db_error) {
                return Err(error);
            }
        }

        attempts += 1;
        if attempts >= MAX_ATTEMPTS {
            return Err(error);
        }

        // The wait time increases exponentially, starting at 100ms and doubling up to a max of 3.2s.
        let wait_time = min(
            MAX_WAIT_TIME,
            STARTING_WAIT_TIME * u64::pow(2, attempts - 1),
        );

        // The jitter is up to half the wait time
        let jitter: u64 = rng.gen_range(0..wait_time / 2);

        std::thread::sleep(Duration::from_millis(wait_time + jitter));
    }
}

// Check if a database error can be retried
fn error_retryable(error: &postgres::error::DbError) -> bool {
    // LOCK_NOT_AVAILABLE is caused by lock_timeout being exceeded
    matches!(error.code(), &postgres::error::SqlState::LOCK_NOT_AVAILABLE)
}



================================================
FILE: src/helpers.rs
================================================
use anyhow::Context;

use crate::db::Conn;

pub fn set_up_helpers(db: &mut dyn Conn, target_migration: &str) -> anyhow::Result<()> {
    let query = format!(
        "
			CREATE OR REPLACE FUNCTION reshape.is_new_schema()
			RETURNS BOOLEAN AS $$
            DECLARE
                setting TEXT := current_setting('reshape.is_new_schema', TRUE);
                setting_bool BOOLEAN := setting IS NOT NULL AND setting = 'YES';
			BEGIN
				RETURN current_setting('search_path') = 'migration_{}' OR setting_bool;
			END
			$$ language 'plpgsql';
        ",
        target_migration,
    );
    db.query(&query)
        .context("failed creating helper function reshape.is_new_schema()")?;

    Ok(())
}

pub fn tear_down_helpers(db: &mut dyn Conn) -> anyhow::Result<()> {
    db.query("DROP FUNCTION IF EXISTS reshape.is_new_schema;")?;
    Ok(())
}



================================================
FILE: src/lib.rs
================================================
use crate::{
    migrations::{Migration, MigrationContext},
    schema::Schema,
};

use anyhow::{anyhow, Context};
use colored::*;
use db::{Conn, DbConn, DbLocker};
use postgres::Config;
use schema::Table;

mod db;
mod helpers;
pub mod migrations;
mod schema;
mod state;

pub use crate::state::State;

pub struct Reshape {
    db: DbLocker,
}

impl Reshape {
    pub fn new(connection_string: &str) -> anyhow::Result<Reshape> {
        let config: Config = connection_string.parse()?;
        Self::new_with_config(&config)
    }

    pub fn new_with_options(
        host: &str,
        port: u16,
        database: &str,
        username: &str,
        password: &str,
    ) -> anyhow::Result<Reshape> {
        let mut config = Config::new();
        config
            .host(host)
            .port(port)
            .user(username)
            .dbname(database)
            .password(password);

        Self::new_with_config(&config)
    }

    fn new_with_config(config: &Config) -> anyhow::Result<Reshape> {
        let db = DbLocker::connect(config)?;
        Ok(Reshape { db })
    }

    pub fn migrate(
        &mut self,
        migrations: impl IntoIterator<Item = Migration>,
    ) -> anyhow::Result<()> {
        self.db.lock(|db| {
            let mut state = State::load(db)?;
            migrate(db, &mut state, migrations)
        })
    }

    pub fn complete(&mut self) -> anyhow::Result<()> {
        self.db.lock(|db| {
            let mut state = State::load(db)?;
            complete(db, &mut state)
        })
    }

    pub fn abort(&mut self) -> anyhow::Result<()> {
        self.db.lock(|db| {
            let mut state = State::load(db)?;
            abort(db, &mut state)
        })
    }

    pub fn remove(&mut self) -> anyhow::Result<()> {
        self.db.lock(|db| {
            let mut state = State::load(db)?;

            // Remove migration schemas and views
            if let Some(current_migration) = &state::current_migration(db)? {
                db.run(&format!(
                    "DROP SCHEMA IF EXISTS {} CASCADE",
                    schema_name_for_migration(current_migration)
                ))?;
            }

            if let State::InProgress { migrations } = &state {
                let target_migration = migrations.last().unwrap().name.to_string();
                db.run(&format!(
                    "DROP SCHEMA IF EXISTS {} CASCADE",
                    schema_name_for_migration(&target_migration)
                ))?;
            }

            // Remove all tables
            let schema = Schema::new();
            for table in schema.get_tables(db)? {
                db.run(&format!(
                    r#"
                    DROP TABLE IF EXISTS "{}" CASCADE
                    "#,
                    table.real_name
                ))?;
            }

            // Remove all enums
            let enums: Vec<String> = db
                .query("SELECT typname FROM pg_type WHERE typcategory = 'E'")?
                .iter()
                .map(|row| row.get("typname"))
                .collect();
            for enum_type in enums {
                db.run(&format!("DROP TYPE {}", enum_type))?;
            }

            // Reset state
            state.clear(db)?;

            println!("Reshape and all data has been removed");

            Ok(())
        })
    }
}

pub fn latest_schema_from_migrations(migrations: &[Migration]) -> Option<String> {
    migrations
        .last()
        .map(|migration| schema_name_for_migration(&migration.name))
}

pub fn schema_query_for_migration(migration_name: &str) -> String {
    let schema_name = schema_name_for_migration(migration_name);
    format!("SET search_path TO {}", schema_name)
}

fn schema_name_for_migration(migration_name: &str) -> String {
    format!("migration_{}", migration_name)
}

fn migrate(
    db: &mut DbConn,
    state: &mut State,
    migrations: impl IntoIterator<Item = Migration>,
) -> anyhow::Result<()> {
    // Make sure no migration is in progress
    if let State::InProgress { .. } = &state {
        println!(
            "Migration already in progress, please complete using 'reshape migration complete'"
        );
        return Ok(());
    }

    if let State::Completing { .. } = &state {
        println!(
                    "Migration already in progress and has started completion, please finish using 'reshape migration complete'"
                );
        return Ok(());
    }

    // Determine which migrations need to be applied by comparing the provided migrations
    // with the already applied ones stored in the state. This will throw an error if the
    // two sets of migrations don't agree, for example if a new migration has been added
    // in between two existing ones.
    let remaining_migrations = state::remaining_migrations(db, migrations)?;
    if remaining_migrations.is_empty() {
        println!("No migrations left to apply");
        return Ok(());
    }

    // If we have already started applying some migrations we need to ensure that
    // they are the same ones we want to apply now
    if let State::Applying {
        migrations: existing_migrations,
    } = &state
    {
        if existing_migrations != &remaining_migrations {
            return Err(anyhow!(
                "a previous migration seems to have failed without cleaning up. Please run `reshape migration abort` and then run migrate again."
            ));
        }
    }

    // Move to the "Applying" state which is necessary as we can't run the migrations
    // and state update as a single transaction. If a migration unexpectedly fails without
    // automatically aborting, this state saves us from dangling migrations. It forces the user
    // to either run migrate again (which works as all migrations are idempotent) or abort.
    state.applying(remaining_migrations.clone());
    state.save(db)?;

    println!("Applying {} migrations\n", remaining_migrations.len());

    let target_migration = remaining_migrations.last().unwrap().name.to_string();
    helpers::set_up_helpers(db, &target_migration).context("failed to set up helpers")?;

    let mut new_schema = Schema::new();
    let mut last_migration_index = usize::MAX;
    let mut last_action_index = usize::MAX;
    let mut result: anyhow::Result<()> = Ok(());

    'outer: for (migration_index, migration) in remaining_migrations.iter().enumerate() {
        println!("Migrating '{}':", migration.name);
        last_migration_index = migration_index;

        for (action_index, action) in migration.actions.iter().enumerate() {
            last_action_index = action_index;

            let description = action.describe();
            print!("  + {} ", description);

            let ctx =
                MigrationContext::new(migration_index, action_index, state::current_migration(db)?);
            result = action
                .run(&ctx, db, &new_schema)
                .with_context(|| format!("failed to {}", description));

            if result.is_ok() {
                action.update_schema(&ctx, &mut new_schema);
                println!("{}", "done".green());
            } else {
                println!("{}", "failed".red());
                break 'outer;
            }
        }

        println!();
    }

    // If a migration failed, we abort all the migrations that were applied
    if let Err(err) = result {
        println!("A migration failed, aborting migrations that have already been applied");

        // Set to the Aborting state. This is to ensure that the failed
        // migration is fully aborted and nothing is left dangling.
        // If the abort is interrupted for any reason, the user can try again
        // by running `reshape migration abort`.
        state.aborting(
            remaining_migrations.clone(),
            last_migration_index + 1,
            last_action_index + 1,
        );

        // Abort will only
        abort(db, state)?;

        return Err(err);
    }

    // Create schema and views for migration
    create_schema_for_migration(db, &target_migration, &new_schema)
        .with_context(|| format!("failed to create schema for migration {}", target_migration))?;

    // Update state once migrations have been performed
    state.in_progress(remaining_migrations);
    state.save(db).context("failed to save in-progress state")?;

    println!("Migrations have been applied and the new schema is ready for use:");
    println!(
        "  - Run '{}' from your application to use the latest schema",
        schema_query_for_migration(&target_migration)
    );
    println!(
        "  - Run 'reshape migration complete' once your application has been updated and the previous schema is no longer in use"
    );
    Ok(())
}

fn complete(db: &mut DbConn, state: &mut State) -> anyhow::Result<()> {
    // Make sure a migration is in progress
    let (remaining_migrations, starting_migration_index, starting_action_index) = match state.clone() {
                State::InProgress { migrations } => {
                    // Move into the Completing state. Once in this state,
                    // the migration can't be aborted and must be completed.
                    state.completing(migrations.clone(), 0, 0);
                    state.save(db).context("failed to save state")?;

                    (migrations, 0, 0)
                },
                State::Completing {
                    migrations,
                    current_migration_index,
                    current_action_index
                } => (migrations, current_migration_index, current_action_index),
                State::Aborting { .. } => {
                    return Err(anyhow!("migration been aborted and can't be completed. Please finish using `reshape migration abort`."))
                }
                State::Applying { .. } => {
                    return Err(anyhow!("a previous migration unexpectedly failed. Please run `reshape migrate` to try applying the migration again."))
                }
                State::Idle => {
                    println!("No migration in progress");
                    return Ok(());
                }
            };

    // Remove previous migration's schema
    if let Some(current_migration) = &state::current_migration(db)? {
        db.run(&format!(
            "DROP SCHEMA IF EXISTS {} CASCADE",
            schema_name_for_migration(current_migration)
        ))
        .context("failed to remove previous migration's schema")?;
    }

    for (migration_index, migration) in remaining_migrations.iter().enumerate() {
        // Skip all the migrations which have already been completed
        if migration_index < starting_migration_index {
            continue;
        }

        println!("Completing '{}':", migration.name);

        for (action_index, action) in migration.actions.iter().enumerate() {
            // Skip all actions which have already been completed
            if migration_index == starting_migration_index && action_index < starting_action_index {
                continue;
            }

            let description = action.describe();
            print!("  + {} ", description);

            let ctx =
                MigrationContext::new(migration_index, action_index, state::current_migration(db)?);

            // Update state to indicate that this action has been completed.
            // We won't save this new state until after the action has completed.
            state.completing(
                remaining_migrations.clone(),
                migration_index + 1,
                action_index + 1,
            );

            // This did_save check is necessary because of the borrow checker.
            // The Transaction which might be returned from action.complete
            // contains a mutable reference to self.db. We need the Transaction
            // to be dropped before we can save the state using self.db instead,
            // which we achieve here by limiting the lifetime of the Transaction
            // with a new block.
            let did_save = {
                let result = action
                    .complete(&ctx, db)
                    .with_context(|| format!("failed to complete migration {}", migration.name))
                    .with_context(|| format!("failed to complete action: {}", description));

                let maybe_transaction = match result {
                    Ok(maybe_transaction) => {
                        println!("{}", "done".green());
                        maybe_transaction
                    }
                    Err(e) => {
                        println!("{}", "failed".red());
                        return Err(e);
                    }
                };

                // Update state with which migrations and actions have been completed.
                // Each action can create and return a transaction if they need atomicity.
                // We use this transaction to update the state to ensure the action only completes.
                // once.
                // We want to use a single transaction for each action to keep the length of
                // the transaction as short as possible. Wherever possible, we don't want to
                // use a transaction at all.
                if let Some(mut transaction) = maybe_transaction {
                    state
                        .save(&mut transaction)
                        .context("failed to save state after completing action")?;
                    transaction
                        .commit()
                        .context("failed to commit transaction")?;

                    true
                } else {
                    false
                }
            };

            // If the action didn't return a transaction we save the state normally instead
            if !did_save {
                state
                    .save(db)
                    .context("failed to save state after completing action")?;
            }
        }

        println!();
    }

    // Remove helpers which are no longer in use
    helpers::tear_down_helpers(db).context("failed to tear down helpers")?;

    state
        .complete(db)
        .context("failed to update state as completed")?;

    Ok(())
}

fn abort(db: &mut DbConn, state: &mut State) -> anyhow::Result<()> {
    let (remaining_migrations, last_migration_index, last_action_index) = match state.clone() {
        State::InProgress { migrations } | State::Applying { migrations } => {
            // Set to the Aborting state. Once this is done, the migration has to
            // be fully aborted and can't be completed.
            state.aborting(migrations.clone(), 0, 0);
            state.save(db)?;

            (migrations, usize::MAX, usize::MAX)
        }
        State::Aborting {
            migrations,
            last_migration_index,
            last_action_index,
        } => (migrations, last_migration_index, last_action_index),
        State::Completing { .. } => {
            return Err(anyhow!("migration completion has already been started. Please run `reshape migration complete` again to finish it."));
        }
        State::Idle => {
            println!("No migration is in progress");
            return Ok(());
        }
    };

    // Remove new migration's schema
    let target_migration = remaining_migrations.last().unwrap().name.to_string();
    let schema_name = schema_name_for_migration(&target_migration);
    db.run(&format!("DROP SCHEMA IF EXISTS {} CASCADE", schema_name,))
        .with_context(|| format!("failed to drop schema {}", schema_name))?;

    // Abort all pending migrations
    // Abort all migrations in reverse order
    for (migration_index, migration) in remaining_migrations.iter().enumerate().rev() {
        // Skip migrations which shouldn't be aborted
        // The reason can be that they have already been aborted or that
        // the migration was never applied in the first place.
        if migration_index >= last_migration_index {
            continue;
        }

        print!("Aborting '{}' ", migration.name);

        for (action_index, action) in migration.actions.iter().enumerate().rev() {
            // Skip actions which shouldn't be aborted
            // The reason can be that they have already been aborted or that
            // the action was never applied in the first place.
            if migration_index == last_migration_index - 1 && action_index >= last_action_index {
                continue;
            }

            let ctx =
                MigrationContext::new(migration_index, action_index, state::current_migration(db)?);
            action
                .abort(&ctx, db)
                .with_context(|| format!("failed to abort migration {}", migration.name))
                .with_context(|| format!("failed to abort action: {}", action.describe()))?;

            // Update state with which migrations and actions have been aborted.
            // We don't need to run this in a transaction as aborts are idempotent.
            state.aborting(remaining_migrations.to_vec(), migration_index, action_index);
            state.save(db).context("failed to save state")?;
        }

        println!("{}", "done".green());
    }

    helpers::tear_down_helpers(db).context("failed to tear down helpers")?;

    *state = State::Idle;
    state.save(db).context("failed to save state")?;

    Ok(())
}

fn create_schema_for_migration(
    db: &mut DbConn,
    migration_name: &str,
    schema: &Schema,
) -> anyhow::Result<()> {
    // Create schema for migration
    let schema_name = schema_name_for_migration(migration_name);
    db.run(&format!("CREATE SCHEMA IF NOT EXISTS {}", schema_name))
        .with_context(|| {
            format!(
                "failed to create schema {} for migration {}",
                schema_name, migration_name
            )
        })?;

    // Create views inside schema
    for table in schema.get_tables(db)? {
        create_view_for_table(db, &table, &schema_name)?;
    }

    Ok(())
}

fn create_view_for_table(db: &mut impl Conn, table: &Table, schema: &str) -> anyhow::Result<()> {
    let select_columns: Vec<String> = table
        .columns
        .iter()
        .map(|column| {
            format!(
                r#"
                    "{real_name}" AS "{alias}"
                    "#,
                real_name = column.real_name,
                alias = column.name,
            )
        })
        .collect();

    db.run(&format!(
        r#"
        CREATE OR REPLACE VIEW {schema}."{view_name}" AS
            SELECT {columns}
            FROM "{table_name}"
        "#,
        schema = schema,
        table_name = table.real_name,
        view_name = table.name,
        columns = select_columns.join(","),
    ))
    .with_context(|| format!("failed to create view for table {}", table.name))?;

    Ok(())
}



================================================
FILE: src/main.rs
================================================
use std::{
    fs::{self, File},
    io::Read,
    path::Path,
};

use anyhow::Context;
use clap::{Args, Parser};
use reshape::{
    migrations::{Action, Migration},
    Reshape,
};
use serde::{Deserialize, Serialize};

#[derive(Parser)]
#[clap(name = "Reshape", version, about)]
struct Opts {
    #[clap(subcommand)]
    cmd: Command,
}

#[derive(Parser)]
#[clap(about)]
enum Command {
    #[clap(subcommand)]
    Migration(MigrationCommand),

    #[clap(
        about = "Output the query your application should use to select the right schema",
        display_order = 2
    )]
    SchemaQuery(FindMigrationsOptions),

    #[clap(
        about = "Deprecated. Use `reshape schema-query` instead",
        display_order = 3
    )]
    GenerateSchemaQuery(FindMigrationsOptions),

    #[clap(
        about = "Deprecated. Use `reshape migration start` instead",
        display_order = 4
    )]
    Migrate(MigrateOptions),
    #[clap(
        about = "Deprecated. Use `reshape migration complete` instead",
        display_order = 5
    )]
    Complete(ConnectionOptions),
    #[clap(
        about = "Deprecated. Use `reshape migration abort` instead",
        display_order = 6
    )]
    Abort(ConnectionOptions),
}

#[derive(Parser)]
#[clap(about = "Commands for managing migrations", display_order = 1)]
enum MigrationCommand {
    #[clap(
        about = "Starts a new migration, applying any migrations which haven't yet been applied",
        display_order = 1
    )]
    Start(MigrateOptions),

    #[clap(about = "Completes an in-progress migration", display_order = 2)]
    Complete(ConnectionOptions),

    #[clap(
        about = "Aborts an in-progress migration without losing any data",
        display_order = 3
    )]
    Abort(ConnectionOptions),
}

#[derive(Args)]
struct MigrateOptions {
    // Some comment
    #[clap(long, short)]
    complete: bool,
    #[clap(flatten)]
    connection_options: ConnectionOptions,
    #[clap(flatten)]
    find_migrations_options: FindMigrationsOptions,
}

#[derive(Parser)]
struct ConnectionOptions {
    #[clap(long)]
    url: Option<String>,
    #[clap(long, default_value = "localhost")]
    host: String,
    #[clap(long, default_value = "5432")]
    port: u16,
    #[clap(long, short, default_value = "postgres")]
    database: String,
    #[clap(long, short, default_value = "postgres")]
    username: String,
    #[clap(long, short, default_value = "postgres")]
    password: String,
}

#[derive(Parser)]
struct FindMigrationsOptions {
    #[clap(long, default_value = "migrations")]
    dirs: Vec<String>,
}

fn main() -> anyhow::Result<()> {
    let opts: Opts = Opts::parse();
    run(opts)
}

fn run(opts: Opts) -> anyhow::Result<()> {
    match opts.cmd {
        Command::Migration(MigrationCommand::Start(opts)) | Command::Migrate(opts) => {
            let mut reshape = reshape_from_connection_options(&opts.connection_options)?;
            let migrations = find_migrations(&opts.find_migrations_options)?;
            reshape.migrate(migrations)?;

            // Automatically complete migration if --complete flag is set
            if opts.complete {
                reshape.complete()?;
            }

            Ok(())
        }
        Command::Migration(MigrationCommand::Complete(opts)) | Command::Complete(opts) => {
            let mut reshape = reshape_from_connection_options(&opts)?;
            reshape.complete()
        }
        Command::Migration(MigrationCommand::Abort(opts)) | Command::Abort(opts) => {
            let mut reshape = reshape_from_connection_options(&opts)?;
            reshape.abort()
        }
        Command::SchemaQuery(opts) | Command::GenerateSchemaQuery(opts) => {
            let migrations = find_migrations(&opts)?;
            let query = migrations
                .last()
                .map(|migration| reshape::schema_query_for_migration(&migration.name));
            println!("{}", query.unwrap_or_else(|| "".to_string()));

            Ok(())
        }
    }
}

fn reshape_from_connection_options(opts: &ConnectionOptions) -> anyhow::Result<Reshape> {
    // Load environment variables from .env file if it exists
    dotenv::dotenv().ok();

    let url_env = std::env::var("DB_URL").ok();
    let url = url_env.as_ref().or_else(|| opts.url.as_ref());

    // Use the connection URL if it has been set
    if let Some(url) = url {
        return Reshape::new(url);
    }

    let host_env = std::env::var("DB_HOST").ok();
    let host = host_env.as_ref().unwrap_or_else(|| &opts.host);

    let port = std::env::var("DB_PORT")
        .ok()
        .and_then(|port| port.parse::<u16>().ok())
        .unwrap_or(opts.port);

    let username_env = std::env::var("DB_USERNAME").ok();
    let username = username_env.as_ref().unwrap_or_else(|| &opts.username);

    let password_env = std::env::var("DB_PASSWORD").ok();
    let password = password_env.as_ref().unwrap_or_else(|| &opts.password);

    let database_env = std::env::var("DB_NAME").ok();
    let database = database_env.as_ref().unwrap_or_else(|| &opts.database);

    Reshape::new_with_options(host, port, database, username, password)
}

fn find_migrations(opts: &FindMigrationsOptions) -> anyhow::Result<Vec<Migration>> {
    let search_paths = opts
        .dirs
        .iter()
        .map(Path::new)
        // Filter out all directories that don't exist
        .filter(|path| path.exists());

    // Find all files in the search paths
    let mut file_paths = Vec::new();
    for search_path in search_paths {
        let entries = fs::read_dir(search_path)?;
        for entry in entries {
            let path = entry?.path();
            file_paths.push(path);
        }
    }

    // Sort all files by their file names (without extension)
    // The files are sorted naturally, e.g. "1_test_migration" < "10_test_migration"
    file_paths.sort_unstable_by(|path1, path2| {
        let file1 = path1.as_path().file_stem().unwrap().to_str().unwrap();
        let file2 = path2.as_path().file_stem().unwrap().to_str().unwrap();

        lexical_sort::natural_cmp(file1, file2)
    });

    file_paths
        .iter()
        .map(|path| {
            let mut file = File::open(path)?;

            // Read file data
            let mut data = String::new();
            file.read_to_string(&mut data)?;

            Ok((path, data))
        })
        .map(|result| {
            result.and_then(|(path, data)| {
                let extension = path.extension().and_then(|ext| ext.to_str()).unwrap();
                let file_migration =
                    decode_migration_file(&data, extension).with_context(|| {
                        format!("failed to parse migration file {}", path.display())
                    })?;

                let file_name = path.file_stem().and_then(|name| name.to_str()).unwrap();
                Ok(Migration {
                    name: file_migration.name.unwrap_or_else(|| file_name.to_string()),
                    description: file_migration.description,
                    actions: file_migration.actions,
                })
            })
        })
        .collect()
}

fn decode_migration_file(data: &str, extension: &str) -> anyhow::Result<FileMigration> {
    let migration: FileMigration = match extension {
        "json" => serde_json::from_str(data)?,
        "toml" => toml::from_str(data)?,
        extension => {
            return Err(anyhow::anyhow!(
                "unrecognized file extension '{}'",
                extension
            ))
        }
    };

    Ok(migration)
}

#[derive(Serialize, Deserialize)]
struct FileMigration {
    name: Option<String>,
    description: Option<String>,
    actions: Vec<Box<dyn Action>>,
}



================================================
FILE: src/schema.rs
================================================
use crate::db::Conn;
use std::collections::{HashMap, HashSet};

// Schema tracks changes made to tables and columns during a migration.
// These changes are not applied until the migration is completed but
// need to be taken into consideration when creating views for a migration
// and when a user references a table or column in a migration.
//
// The changes to a table are tracked by a `TableChanges` struct. The possible
// changes are:
//   - Changing the name which updates `current_name`.
//   - Removing which sets the `removed` flag.
//
// Changes to a column are tracked by a `ColumnChanges` struct which reside in
// the corresponding `TableChanges`. The possible changes are:
//   - Changing the name which updates `current_name`.
//   - Changing the backing column which will add the new column to the end of
//     `intermediate_columns`. This is used when temporary columns are
//     introduced which will eventually replace the current column.
//   - Removing which sets the `removed` flag.
//
// Schema provides some schema introspection methods, `get_tables` and `get_table`,
// which will retrieve the current schema from the database and apply the changes.
#[derive(Debug)]
pub struct Schema {
    table_changes: Vec<TableChanges>,
}

impl Schema {
    pub fn new() -> Schema {
        Schema {
            table_changes: Vec::new(),
        }
    }

    pub fn change_table<F>(&mut self, current_name: &str, f: F)
    where
        F: FnOnce(&mut TableChanges),
    {
        let table_change_index = self
            .table_changes
            .iter()
            .position(|table| table.current_name == current_name)
            .unwrap_or_else(|| {
                let new_changes = TableChanges::new(current_name.to_string());
                self.table_changes.push(new_changes);
                self.table_changes.len() - 1
            });

        let table_changes = &mut self.table_changes[table_change_index];
        f(table_changes)
    }
}

impl Default for Schema {
    fn default() -> Self {
        Self::new()
    }
}

#[derive(Debug)]
pub struct TableChanges {
    current_name: String,
    real_name: String,
    column_changes: Vec<ColumnChanges>,
    removed: bool,
}

impl TableChanges {
    fn new(name: String) -> Self {
        Self {
            current_name: name.to_string(),
            real_name: name,
            column_changes: Vec::new(),
            removed: false,
        }
    }

    pub fn set_name(&mut self, name: &str) {
        self.current_name = name.to_string();
    }

    pub fn change_column<F>(&mut self, current_name: &str, f: F)
    where
        F: FnOnce(&mut ColumnChanges),
    {
        let column_change_index = self
            .column_changes
            .iter()
            .position(|column| column.current_name == current_name)
            .unwrap_or_else(|| {
                let new_changes = ColumnChanges::new(current_name.to_string());
                self.column_changes.push(new_changes);
                self.column_changes.len() - 1
            });

        let column_changes = &mut self.column_changes[column_change_index];
        f(column_changes)
    }

    pub fn set_removed(&mut self) {
        self.removed = true;
    }
}

#[derive(Debug)]
pub struct ColumnChanges {
    current_name: String,
    backing_columns: Vec<String>,
    removed: bool,
}

impl ColumnChanges {
    fn new(name: String) -> Self {
        Self {
            current_name: name.to_string(),
            backing_columns: vec![name],
            removed: false,
        }
    }

    pub fn set_name(&mut self, name: &str) {
        self.current_name = name.to_string();
    }

    pub fn set_column(&mut self, column_name: &str) {
        self.backing_columns.push(column_name.to_string())
    }

    pub fn set_removed(&mut self) {
        self.removed = true;
    }

    fn real_name(&self) -> &str {
        self.backing_columns
            .last()
            .expect("backing_columns should never be empty")
    }
}

#[derive(Debug)]
pub struct Table {
    pub name: String,
    pub real_name: String,
    pub columns: Vec<Column>,
}

#[derive(Debug)]
pub struct Column {
    pub name: String,
    pub real_name: String,
    pub data_type: String,
    pub nullable: bool,
    pub default: Option<String>,
}

impl Schema {
    pub fn get_tables(&self, db: &mut dyn Conn) -> anyhow::Result<Vec<Table>> {
        db.query(
            "
            SELECT table_name
            FROM information_schema.tables
            WHERE table_schema = 'public'
            ",
        )?
        .iter()
        .map(|row| row.get::<'_, _, String>("table_name"))
        .filter_map(|real_name| {
            let table_changes = self
                .table_changes
                .iter()
                .find(|changes| changes.real_name == real_name);

            // Skip table if it has been removed
            if let Some(changes) = table_changes {
                if changes.removed {
                    return None;
                }
            }

            Some(self.get_table_by_real_name(db, &real_name))
        })
        .collect()
    }

    pub fn get_table(&self, db: &mut dyn Conn, table_name: &str) -> anyhow::Result<Table> {
        let table_changes = self
            .table_changes
            .iter()
            .find(|changes| changes.current_name == table_name);

        let real_table_name = table_changes
            .map(|changes| changes.real_name.to_string())
            .unwrap_or_else(|| table_name.to_string());

        self.get_table_by_real_name(db, &real_table_name)
    }

    fn get_table_by_real_name(
        &self,
        db: &mut dyn Conn,
        real_table_name: &str,
    ) -> anyhow::Result<Table> {
        let table_changes = self
            .table_changes
            .iter()
            .find(|changes| changes.real_name == real_table_name);

        let real_columns: Vec<(String, String, bool, Option<String>)> = db
            .query(&format!(
                "
                SELECT column_name, CASE WHEN data_type = 'USER-DEFINED' THEN udt_name ELSE data_type END, is_nullable, column_default
                FROM information_schema.columns
                WHERE table_name = '{table}' AND table_schema = 'public'
                ORDER BY ordinal_position
                ",
                table = real_table_name,
            ))?
            .iter()
            .map(|row| {
                (
                    row.get("column_name"),
                    row.get("data_type"),
                    row.get::<'_, _, String>("is_nullable") == "YES",
                    row.get("column_default"),
                )
            })
            .collect();

        let mut ignore_columns: HashSet<String> = HashSet::new();
        let mut aliases: HashMap<String, &str> = HashMap::new();

        if let Some(changes) = table_changes {
            for column_changes in &changes.column_changes {
                if column_changes.removed {
                    ignore_columns.insert(column_changes.real_name().to_string());
                } else {
                    aliases.insert(
                        column_changes.real_name().to_string(),
                        &column_changes.current_name,
                    );
                }

                let (_, rest) = column_changes
                    .backing_columns
                    .split_last()
                    .expect("backing_columns should never be empty");

                for column in rest {
                    ignore_columns.insert(column.to_string());
                }
            }
        }

        let mut columns: Vec<Column> = Vec::new();

        for (real_name, data_type, nullable, default) in real_columns {
            if ignore_columns.contains(&*real_name) {
                continue;
            }

            let name = aliases
                .get(&real_name)
                .map(|alias| alias.to_string())
                .unwrap_or_else(|| real_name.to_string());

            columns.push(Column {
                name,
                real_name,
                data_type,
                nullable,
                default,
            });
        }

        let current_table_name = table_changes
            .map(|changes| changes.current_name.as_ref())
            .unwrap_or_else(|| real_table_name);

        let table = Table {
            name: current_table_name.to_string(),
            real_name: real_table_name.to_string(),
            columns,
        };

        Ok(table)
    }
}

impl Table {
    pub fn real_column_names<'a>(
        &'a self,
        columns: &'a [String],
    ) -> impl Iterator<Item = &'a String> {
        columns.iter().map(|name| {
            self.get_column(name)
                .map(|col| &col.real_name)
                .unwrap_or(name)
        })
    }

    pub fn get_column(&self, name: &str) -> Option<&Column> {
        self.columns.iter().find(|column| column.name == name)
    }
}



================================================
FILE: src/state.rs
================================================
use crate::{db::Conn, migrations::Migration};
use anyhow::anyhow;

use serde::{Deserialize, Serialize};
use version::version;

#[derive(Serialize, Deserialize, Clone, Debug)]
#[serde(tag = "state")]
pub enum State {
    #[serde(rename = "idle")]
    Idle,

    #[serde(rename = "applying")]
    Applying { migrations: Vec<Migration> },

    #[serde(rename = "in_progress")]
    InProgress { migrations: Vec<Migration> },

    #[serde(rename = "completing")]
    Completing {
        migrations: Vec<Migration>,
        current_migration_index: usize,
        current_action_index: usize,
    },

    #[serde(rename = "aborting")]
    Aborting {
        migrations: Vec<Migration>,
        last_migration_index: usize,
        last_action_index: usize,
    },
}

impl State {
    pub fn load(db: &mut impl Conn) -> anyhow::Result<State> {
        Self::ensure_schema_and_table(db)?;

        let results = db.query("SELECT value FROM reshape.data WHERE key = 'state'")?;

        let state = match results.first() {
            Some(row) => {
                let json: serde_json::Value = row.get(0);
                serde_json::from_value(json)?
            }
            None => Default::default(),
        };
        Ok(state)
    }

    pub fn save(&self, db: &mut impl Conn) -> anyhow::Result<()> {
        Self::ensure_schema_and_table(db)?;

        let json = serde_json::to_value(self)?;
        db.query_with_params(
            "INSERT INTO reshape.data (key, value) VALUES ('state', $1) ON CONFLICT (key) DO UPDATE SET value = $1",
            &[&json]
        )?;
        Ok(())
    }

    pub fn clear(&mut self, db: &mut impl Conn) -> anyhow::Result<()> {
        db.run("DROP SCHEMA reshape CASCADE")?;

        *self = Self::default();

        Ok(())
    }

    // Complete will change the state from Completing to Idle
    pub fn complete(&mut self, db: &mut impl Conn) -> anyhow::Result<()> {
        let current_state = std::mem::replace(self, Self::Idle);

        match current_state {
            Self::Completing { migrations, .. } => {
                // Add migrations and update state in a transaction to ensure atomicity
                let mut transaction = db.transaction()?;
                save_migrations(&mut transaction, &migrations)?;
                self.save(&mut transaction)?;
                transaction.commit()?;
            }
            _ => {
                // Move old state back
                *self = current_state;

                return Err(anyhow!(
                    "couldn't update state to be completed, not in Completing state"
                ));
            }
        }

        Ok(())
    }

    pub fn applying(&mut self, new_migrations: Vec<Migration>) {
        *self = Self::Applying {
            migrations: new_migrations,
        };
    }

    pub fn in_progress(&mut self, new_migrations: Vec<Migration>) {
        *self = Self::InProgress {
            migrations: new_migrations,
        };
    }

    pub fn completing(
        &mut self,
        migrations: Vec<Migration>,
        current_migration_index: usize,
        current_action_index: usize,
    ) {
        *self = Self::Completing {
            migrations,
            current_migration_index,
            current_action_index,
        }
    }

    pub fn aborting(
        &mut self,
        migrations: Vec<Migration>,
        last_migration_index: usize,
        last_action_index: usize,
    ) {
        *self = Self::Aborting {
            migrations,
            last_migration_index,
            last_action_index,
        }
    }

    fn ensure_schema_and_table(db: &mut impl Conn) -> anyhow::Result<()> {
        db.run("CREATE SCHEMA IF NOT EXISTS reshape")?;

        // Create data table which will be a key-value table containing
        // the version and current state.
        db.run("CREATE TABLE IF NOT EXISTS reshape.data (key TEXT PRIMARY KEY, value JSONB)")?;

        // Create migrations table which will store all completed migrations
        db.run(
            "
            CREATE TABLE IF NOT EXISTS reshape.migrations (
                index INTEGER GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
                name TEXT NOT NULL,
                description TEXT,
                actions JSONB NOT NULL,
                completed_at TIMESTAMP DEFAULT NOW()
            )
            ",
        )?;

        // Update the current version
        let encoded_version = serde_json::to_value(version!().to_string())?;
        db.query_with_params(
            "
            INSERT INTO reshape.data (key, value)
            VALUES ('version', $1)
            ON CONFLICT (key) DO UPDATE SET value = $1
            ",
            &[&encoded_version],
        )?;

        Ok(())
    }
}

impl Default for State {
    fn default() -> Self {
        Self::Idle
    }
}

pub fn current_migration(db: &mut dyn Conn) -> anyhow::Result<Option<String>> {
    let name: Option<String> = db
        .query(
            "
            SELECT name
            FROM reshape.migrations
            ORDER BY index DESC
            LIMIT 1
            ",
        )?
        .first()
        .map(|row| row.get("name"));
    Ok(name)
}

pub fn remaining_migrations(
    db: &mut impl Conn,
    new_migrations: impl IntoIterator<Item = Migration>,
) -> anyhow::Result<Vec<Migration>> {
    let mut new_iter = new_migrations.into_iter();

    // Ensure the new migrations match up with the existing ones
    let mut highest_index: Option<i32> = None;
    loop {
        let migrations = get_migrations(db, highest_index)?;
        if migrations.is_empty() {
            break;
        }

        for (index, existing) in migrations {
            highest_index = Some(index);

            let new = match new_iter.next() {
                Some(migration) => migration,
                None => {
                    return Err(anyhow!(
                        "existing migration {} doesn't exist in local migrations",
                        existing
                    ))
                }
            };

            if existing != new.name {
                return Err(anyhow!(
                    "existing migration {} does not match new migration {}",
                    existing,
                    new.name
                ));
            }
        }
    }

    // Return the remaining migrations
    let items: Vec<Migration> = new_iter.collect();
    Ok(items)
}

fn get_migrations(
    db: &mut impl Conn,
    index_larger_than: Option<i32>,
) -> anyhow::Result<Vec<(i32, String)>> {
    let rows = if let Some(index_larger_than) = index_larger_than {
        db.query_with_params(
            "
            SELECT index, name
            FROM reshape.migrations
            WHERE index > $1
            ORDER BY index ASC
            LIMIT 100
            ",
            &[&index_larger_than],
        )?
    } else {
        db.query(
            "
            SELECT index, name
            FROM reshape.migrations
            LIMIT 100
            ",
        )?
    };

    let migrations = rows
        .iter()
        .map(|row| (row.get("index"), row.get("name")))
        .collect();
    Ok(migrations)
}

fn save_migrations(db: &mut impl Conn, migrations: &[Migration]) -> anyhow::Result<()> {
    for migration in migrations {
        let encoded_actions = serde_json::to_value(&migration.actions)?;
        db.query_with_params(
            "INSERT INTO reshape.migrations(name, description, actions) VALUES ($1, $2, $3)",
            &[&migration.name, &migration.description, &encoded_actions],
        )?;
    }

    Ok(())
}



================================================
FILE: src/migrations/add_column.rs
================================================
use super::{common, Action, Column, MigrationContext};
use crate::{
    db::{Conn, Transaction},
    schema::Schema,
};
use anyhow::{bail, Context};
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize, Debug)]
pub struct AddColumn {
    pub table: String,
    pub column: Column,
    pub up: Option<Transformation>,
}

#[derive(Serialize, Deserialize, Debug)]
#[serde(untagged)]
pub enum Transformation {
    Simple(String),
    Update {
        table: String,
        value: String,
        r#where: String,
    },
}

impl AddColumn {
    fn temp_column_name(&self, ctx: &MigrationContext) -> String {
        format!(
            "{}_temp_column_{}_{}",
            ctx.prefix(),
            self.table,
            self.column.name,
        )
    }

    fn trigger_name(&self, ctx: &MigrationContext) -> String {
        format!(
            "{}_add_column_{}_{}",
            ctx.prefix(),
            self.table,
            self.column.name
        )
    }

    fn reverse_trigger_name(&self, ctx: &MigrationContext) -> String {
        format!(
            "{}_add_column_{}_{}_rev",
            ctx.prefix(),
            self.table,
            self.column.name
        )
    }

    fn not_null_constraint_name(&self, ctx: &MigrationContext) -> String {
        format!(
            "{}_add_column_not_null_{}_{}",
            ctx.prefix(),
            self.table,
            self.column.name
        )
    }
}

#[typetag::serde(name = "add_column")]
impl Action for AddColumn {
    fn describe(&self) -> String {
        format!(
            "Adding column \"{}\" to \"{}\"",
            self.column.name, self.table
        )
    }

    fn run(
        &self,
        ctx: &MigrationContext,
        db: &mut dyn Conn,
        schema: &Schema,
    ) -> anyhow::Result<()> {
        let table = schema.get_table(db, &self.table)?;
        let temp_column_name = self.temp_column_name(ctx);

        let mut definition_parts = vec![
            format!("\"{}\"", temp_column_name.to_string()),
            self.column.data_type.to_string(),
        ];

        if let Some(default) = &self.column.default {
            definition_parts.push("DEFAULT".to_string());
            definition_parts.push(default.to_string());
        }

        if let Some(generated) = &self.column.generated {
            definition_parts.push("GENERATED".to_string());
            definition_parts.push(generated.to_string());
        }

        // Add column as NOT NULL
        let query = format!(
            r#"
			ALTER TABLE "{table}"
            ADD COLUMN IF NOT EXISTS {definition};
			"#,
            table = self.table,
            definition = definition_parts.join(" "),
        );
        db.run(&query).context("failed to add column")?;

        let declarations: Vec<String> = table
            .columns
            .iter()
            .map(|column| {
                format!(
                    "\"{alias}\" public.{table}.{real_name}%TYPE := NEW.{real_name};",
                    table = table.real_name,
                    alias = column.name,
                    real_name = column.real_name,
                )
            })
            .collect();

        if let Some(up) = &self.up {
            if let Transformation::Simple(up) = up {
                // Add triggers to fill in values as they are inserted/updated
                let query = format!(
                    r#"
                    CREATE OR REPLACE FUNCTION {trigger_name}()
                    RETURNS TRIGGER AS $$
                    BEGIN
                        IF NOT reshape.is_new_schema() THEN
                            DECLARE
                                {declarations}
                            BEGIN
                                NEW."{temp_column_name}" = {up};
                            END;
                        END IF;
                        RETURN NEW;
                    END
                    $$ language 'plpgsql';

                    DROP TRIGGER IF EXISTS "{trigger_name}" ON "{table}";
                    CREATE TRIGGER "{trigger_name}" BEFORE UPDATE OR INSERT ON "{table}" FOR EACH ROW EXECUTE PROCEDURE {trigger_name}();
                    "#,
                    temp_column_name = temp_column_name,
                    trigger_name = self.trigger_name(ctx),
                    up = up,
                    table = self.table,
                    declarations = declarations.join("\n"),
                );
                db.run(&query).context("failed to create up trigger")?;

                // Backfill values in batches
                common::batch_touch_rows(db, &table.real_name, Some(&temp_column_name))
                    .context("failed to batch update existing rows")?;
            }

            if let Transformation::Update {
                table: from_table,
                value,
                r#where,
            } = up
            {
                let existing_schema_name = match &ctx.existing_schema_name {
                    Some(name) => name,
                    None => bail!("can't use update without previous migration"),
                };

                let from_table = schema.get_table(db, &from_table)?;

                let from_table_assignments: Vec<String> = from_table
                    .columns
                    .iter()
                    .map(|column| {
                        format!(
                            "{table}.{alias} = NEW.{real_name};",
                            table = from_table.name,
                            alias = column.name,
                            real_name = column.real_name,
                        )
                    })
                    .collect();

                // Add triggers to fill in values as they are inserted/updated
                let query = format!(
                    r#"
                    CREATE OR REPLACE FUNCTION {trigger_name}()
                    RETURNS TRIGGER AS $$
                    #variable_conflict use_variable
                    BEGIN
                        IF NOT reshape.is_new_schema() THEN
                            DECLARE
                                {from_table} migration_{existing_schema_name}.{from_table}%ROWTYPE;
                            BEGIN
                                {assignments}

                                -- Don't trigger reverse trigger when making this update
                                perform set_config('reshape.disable_triggers', 'TRUE', TRUE);

                                UPDATE public."{changed_table_real}"
                                SET "{temp_column_name}" = {value}
                                WHERE {where};

                                perform set_config('reshape.disable_triggers', '', TRUE);
                            END;
                        END IF;
                        RETURN NEW;
                    END
                    $$ language 'plpgsql';

                    DROP TRIGGER IF EXISTS "{trigger_name}" ON "{from_table_real}";
                    CREATE TRIGGER "{trigger_name}" BEFORE UPDATE OR INSERT ON "{from_table_real}" FOR EACH ROW EXECUTE PROCEDURE {trigger_name}();
                    "#,
                    assignments = from_table_assignments.join("\n"),
                    changed_table_real = table.real_name,
                    from_table = from_table.name,
                    from_table_real = from_table.real_name,
                    trigger_name = self.trigger_name(ctx),
                    // declarations = from_table_declarations.join("\n"),
                    temp_column_name = temp_column_name,
                );
                db.run(&query).context("failed to create up trigger")?;

                let from_table_columns = from_table
                    .columns
                    .iter()
                    .map(|column| format!("{} as {}", column.real_name, column.name))
                    .collect::<Vec<String>>()
                    .join(", ");

                let changed_table_assignments: Vec<String> = table
                    .columns
                    .iter()
                    .map(|column| {
                        format!(
                            "{table}.{alias} := NEW.{real_name};",
                            table = table.name,
                            alias = column.name,
                            real_name = column.real_name,
                        )
                    })
                    .collect();

                // Add triggers to fill in values as they are inserted/updated
                let query = format!(
                    r#"
                    CREATE OR REPLACE FUNCTION {trigger_name}()
                    RETURNS TRIGGER AS $$
                    #variable_conflict use_variable
                    BEGIN
                        IF NOT reshape.is_new_schema() AND NOT current_setting('reshape.disable_triggers', TRUE) = 'TRUE' THEN
                            DECLARE
                                {changed_table} migration_{existing_schema_name}.{changed_table}%ROWTYPE;
                                __temp_row migration_{existing_schema_name}.{from_table}%ROWTYPE;
                            BEGIN
                                {changed_table_assignments}

                                SELECT {from_table_columns}
                                INTO __temp_row
                                FROM migration_{existing_schema_name}.{from_table} {from_table}
                                WHERE {where};

                                DECLARE
                                    {from_table} migration_{existing_schema_name}.{from_table}%ROWTYPE;
                                BEGIN
                                    {from_table} = __temp_row;
                                    NEW.{temp_column_name} = {value};
                                END;
                            END;
                        END IF;
                        RETURN NEW;
                    END
                    $$ language 'plpgsql';

                    DROP TRIGGER IF EXISTS "{trigger_name}" ON "{changed_table_real}";
                    CREATE TRIGGER "{trigger_name}" BEFORE UPDATE OR INSERT ON "{changed_table_real}" FOR EACH ROW EXECUTE PROCEDURE {trigger_name}();
                    "#,
                    changed_table_assignments = changed_table_assignments.join("\n"),
                    changed_table_real = table.real_name,
                    changed_table = table.name,
                    from_table = from_table.name,
                    trigger_name = self.reverse_trigger_name(ctx),
                    temp_column_name = temp_column_name,
                    // declarations = declarations.join("\n"),
                );
                db.run(&query)
                    .context("failed to create reverse up trigger")?;

                // Backfill values in batches by touching the from table
                common::batch_touch_rows(db, &from_table.real_name, None)
                    .context("failed to batch update existing rows")?;
            }
        }

        // Add a temporary NOT NULL constraint if the column shouldn't be nullable.
        // This constraint is set as NOT VALID so it doesn't apply to existing rows and
        // the existing rows don't need to be scanned under an exclusive lock.
        // Thanks to this, we can set the full column as NOT NULL later with minimal locking.
        if !self.column.nullable {
            let query = format!(
                r#"
                 ALTER TABLE "{table}"
                 ADD CONSTRAINT "{constraint_name}"
                 CHECK ("{column}" IS NOT NULL) NOT VALID
                 "#,
                table = self.table,
                constraint_name = self.not_null_constraint_name(ctx),
                column = temp_column_name,
            );
            db.run(&query)
                .context("failed to add NOT NULL constraint")?;
        }

        Ok(())
    }

    fn complete<'a>(
        &self,
        ctx: &MigrationContext,
        db: &'a mut dyn Conn,
    ) -> anyhow::Result<Option<Transaction<'a>>> {
        let mut transaction = db.transaction().context("failed to create transaction")?;

        // Remove triggers and procedures
        let query = format!(
            r#"
            DROP FUNCTION IF EXISTS "{trigger_name}" CASCADE;
            DROP FUNCTION IF EXISTS "{reverse_trigger_name}" CASCADE;
            "#,
            trigger_name = self.trigger_name(ctx),
            reverse_trigger_name = self.reverse_trigger_name(ctx),
        );
        transaction
            .run(&query)
            .context("failed to drop up trigger")?;

        // Update column to be NOT NULL if necessary
        if !self.column.nullable {
            // Validate the temporary constraint (should always be valid).
            // This performs a sequential scan but does not take an exclusive lock.
            let query = format!(
                r#"
                ALTER TABLE "{table}"
                VALIDATE CONSTRAINT "{constraint_name}"
                "#,
                table = self.table,
                constraint_name = self.not_null_constraint_name(ctx),
            );
            transaction
                .run(&query)
                .context("failed to validate NOT NULL constraint")?;

            // Update the column to be NOT NULL.
            // This requires an exclusive lock but since PG 12 it can check
            // the existing constraint for correctness which makes the lock short-lived.
            // Source: https://dba.stackexchange.com/a/268128
            let query = format!(
                r#"
                ALTER TABLE "{table}"
                ALTER COLUMN "{column}" SET NOT NULL
                "#,
                table = self.table,
                column = self.temp_column_name(ctx),
            );
            transaction
                .run(&query)
                .context("failed to set column as NOT NULL")?;

            // Drop the temporary constraint
            let query = format!(
                r#"
                ALTER TABLE "{table}"
                DROP CONSTRAINT "{constraint_name}"
                "#,
                table = self.table,
                constraint_name = self.not_null_constraint_name(ctx),
            );
            transaction
                .run(&query)
                .context("failed to drop NOT NULL constraint")?;
        }

        // Rename the temporary column to its real name
        transaction
            .run(&format!(
                r#"
                ALTER TABLE "{table}"
                RENAME COLUMN "{temp_column_name}" TO "{column_name}"
                "#,
                table = self.table,
                temp_column_name = self.temp_column_name(ctx),
                column_name = self.column.name,
            ))
            .context("failed to rename column to final name")?;

        Ok(Some(transaction))
    }

    fn update_schema(&self, ctx: &MigrationContext, schema: &mut Schema) {
        schema.change_table(&self.table, |table_changes| {
            table_changes.change_column(&self.column.name, |column_changes| {
                column_changes.set_column(&self.temp_column_name(ctx));
            })
        });
    }

    fn abort(&self, ctx: &MigrationContext, db: &mut dyn Conn) -> anyhow::Result<()> {
        // Remove column
        let query = format!(
            r#"
            ALTER TABLE "{table}"
            DROP COLUMN IF EXISTS "{column}"
            "#,
            table = self.table,
            column = self.temp_column_name(ctx),
        );
        db.run(&query).context("failed to drop column")?;

        // Remove triggers and procedures
        let query = format!(
            r#"
            DROP FUNCTION IF EXISTS "{trigger_name}" CASCADE;
            DROP FUNCTION IF EXISTS "{reverse_trigger_name}" CASCADE;
            "#,
            trigger_name = self.trigger_name(ctx),
            reverse_trigger_name = self.reverse_trigger_name(ctx),
        );
        db.run(&query).context("failed to drop up trigger")?;

        Ok(())
    }
}



================================================
FILE: src/migrations/add_foreign_key.rs
================================================
use super::{common::ForeignKey, Action, MigrationContext};
use crate::{
    db::{Conn, Transaction},
    schema::Schema,
};
use anyhow::Context;
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize, Debug)]
pub struct AddForeignKey {
    pub table: String,
    foreign_key: ForeignKey,
}

#[typetag::serde(name = "add_foreign_key")]
impl Action for AddForeignKey {
    fn describe(&self) -> String {
        format!(
            "Adding foreign key from table \"{}\" to \"{}\"",
            self.table, self.foreign_key.referenced_table
        )
    }

    fn run(
        &self,
        ctx: &MigrationContext,
        db: &mut dyn Conn,
        schema: &Schema,
    ) -> anyhow::Result<()> {
        let table = schema.get_table(db, &self.table)?;
        let referenced_table = schema.get_table(db, &self.foreign_key.referenced_table)?;

        // Add quotes around all column names
        let columns: Vec<String> = table
            .real_column_names(&self.foreign_key.columns)
            .map(|col| format!("\"{}\"", col))
            .collect();
        let referenced_columns: Vec<String> = referenced_table
            .real_column_names(&self.foreign_key.referenced_columns)
            .map(|col| format!("\"{}\"", col))
            .collect();

        // Create foreign key but set is as NOT VALID.
        // This means the foreign key will be enforced for inserts and updates
        // but the existing data won't be checked, that would cause a long-lived lock.
        db.run(&format!(
            r#"
            ALTER TABLE "{table}"
            ADD CONSTRAINT {constraint_name}
            FOREIGN KEY ({columns})
            REFERENCES "{referenced_table}" ({referenced_columns})
            NOT VALID
            "#,
            table = table.real_name,
            constraint_name = self.temp_constraint_name(ctx),
            columns = columns.join(", "),
            referenced_table = referenced_table.real_name,
            referenced_columns = referenced_columns.join(", "),
        ))
        .context("failed to create foreign key")?;

        db.run(&format!(
            r#"
            ALTER TABLE "{table}"
            VALIDATE CONSTRAINT "{constraint_name}"
            "#,
            table = table.real_name,
            constraint_name = self.temp_constraint_name(ctx),
        ))
        .context("failed to validate foreign key")?;

        Ok(())
    }

    fn complete<'a>(
        &self,
        ctx: &MigrationContext,
        db: &'a mut dyn Conn,
    ) -> anyhow::Result<Option<Transaction<'a>>> {
        db.run(&format!(
            r#"
            ALTER TABLE {table}
            RENAME CONSTRAINT {temp_constraint_name} TO {constraint_name}
            "#,
            table = self.table,
            temp_constraint_name = self.temp_constraint_name(ctx),
            constraint_name = self.final_constraint_name(),
        ))
        .context("failed to rename temporary constraint")?;
        Ok(None)
    }

    fn update_schema(&self, _ctx: &MigrationContext, _schema: &mut Schema) {}

    fn abort(&self, ctx: &MigrationContext, db: &mut dyn Conn) -> anyhow::Result<()> {
        db.run(&format!(
            r#"
            ALTER TABLE "{table}"
            DROP CONSTRAINT IF EXISTS "{constraint_name}"
            "#,
            table = self.table,
            constraint_name = self.temp_constraint_name(ctx),
        ))
        .context("failed to validate foreign key")?;

        Ok(())
    }
}

impl AddForeignKey {
    fn temp_constraint_name(&self, ctx: &MigrationContext) -> String {
        format!("{}_temp_fkey", ctx.prefix())
    }

    fn final_constraint_name(&self) -> String {
        format!(
            "{table}_{columns}_fkey",
            table = self.table,
            columns = self.foreign_key.columns.join("_")
        )
    }
}



================================================
FILE: src/migrations/add_index.rs
================================================
use super::{Action, MigrationContext};
use crate::{
    db::{Conn, Transaction},
    schema::Schema,
};
use anyhow::Context;
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize, Debug)]
pub struct AddIndex {
    pub table: String,
    pub index: Index,
}

#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct Index {
    pub name: String,
    pub columns: Vec<String>,
    #[serde(default)]
    pub unique: bool,
    #[serde(rename = "type")]
    pub index_type: Option<String>,
}

#[typetag::serde(name = "add_index")]
impl Action for AddIndex {
    fn describe(&self) -> String {
        format!(
            "Adding index \"{}\" to table \"{}\"",
            self.index.name, self.table
        )
    }

    fn run(
        &self,
        _ctx: &MigrationContext,
        db: &mut dyn Conn,
        schema: &Schema,
    ) -> anyhow::Result<()> {
        let table = schema.get_table(db, &self.table)?;

        let column_real_names: Vec<String> = table
            .columns
            .iter()
            .filter(|column| self.index.columns.contains(&column.name))
            .map(|column| format!("\"{}\"", column.real_name))
            .collect();

        let unique = if self.index.unique { "UNIQUE" } else { "" };
        let index_type_def = if let Some(index_type) = &self.index.index_type {
            format!("USING {index_type}")
        } else {
            "".to_string()
        };

        db.run(&format!(
            r#"
			CREATE {unique} INDEX CONCURRENTLY "{name}" ON "{table}" {index_type_def} ({columns}) 
			"#,
            name = self.index.name,
            table = self.table,
            columns = column_real_names.join(", "),
        ))
        .context("failed to create index")?;
        Ok(())
    }

    fn complete<'a>(
        &self,
        _ctx: &MigrationContext,
        _db: &'a mut dyn Conn,
    ) -> anyhow::Result<Option<Transaction<'a>>> {
        Ok(None)
    }

    fn update_schema(&self, _ctx: &MigrationContext, _schema: &mut Schema) {}

    fn abort(&self, _ctx: &MigrationContext, db: &mut dyn Conn) -> anyhow::Result<()> {
        db.run(&format!(
            r#"
			DROP INDEX CONCURRENTLY IF EXISTS "{name}"
			"#,
            name = self.index.name,
        ))
        .context("failed to drop index")?;
        Ok(())
    }
}



================================================
FILE: src/migrations/alter_column.rs
================================================
use super::{Action, MigrationContext};
use crate::{
    db::{Conn, Transaction},
    migrations::common,
    schema::Schema,
};
use anyhow::{anyhow, Context};
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize, Debug)]
pub struct AlterColumn {
    pub table: String,
    pub column: String,
    pub up: Option<String>,
    pub down: Option<String>,
    #[serde(default)]
    pub changes: ColumnChanges,
}

#[derive(Serialize, Deserialize, Default, Debug)]
pub struct ColumnChanges {
    pub name: Option<String>,
    #[serde(rename = "type")]
    pub data_type: Option<String>,
    pub nullable: Option<bool>,
    pub default: Option<String>,
}

#[typetag::serde(name = "alter_column")]
impl Action for AlterColumn {
    fn describe(&self) -> String {
        format!("Altering column \"{}\" on \"{}\"", self.column, self.table)
    }

    fn run(
        &self,
        ctx: &MigrationContext,
        db: &mut dyn Conn,
        schema: &Schema,
    ) -> anyhow::Result<()> {
        // If we are only changing the name of a column, we don't have to do anything at this stage
        // We'll set the new schema to point to the old column. When the migration is completed,
        // we rename the actual column.
        if self.can_short_circuit() {
            return Ok(());
        }

        let table = schema.get_table(db, &self.table)?;

        let column = table
            .get_column(&self.column)
            .ok_or_else(|| anyhow!("no such column {} exists", self.column))?;

        let temporary_column_name = self.temporary_column_name(ctx);
        let temporary_column_type = self.changes.data_type.as_ref().unwrap_or(&column.data_type);

        // Add temporary, nullable column
        let mut temp_column_definition_parts: Vec<&str> =
            vec![&temporary_column_name, temporary_column_type];

        // Use either new default value or existing one if one exists
        let default_value = self
            .changes
            .default
            .as_ref()
            .or_else(|| column.default.as_ref());
        if let Some(default) = default_value {
            temp_column_definition_parts.push("DEFAULT");
            temp_column_definition_parts.push(default);
        }

        let query = format!(
            r#"
			ALTER TABLE "{table}"
            ADD COLUMN IF NOT EXISTS {temp_column_definition}
			"#,
            table = self.table,
            temp_column_definition = temp_column_definition_parts.join(" "),
        );
        db.run(&query).context("failed to add temporary column")?;

        // If up or down wasn't provided, we default to simply moving the value over.
        // This is the correct behaviour for example when only changing the default value.
        let up = self.up.as_ref().unwrap_or(&self.column);
        let down = self.down.as_ref().unwrap_or(&self.column);

        let declarations: Vec<String> = table
            .columns
            .iter()
            .filter(|column| column.name != self.column)
            .map(|column| {
                format!(
                    "{alias} public.{table}.{real_name}%TYPE := NEW.{real_name};",
                    table = table.real_name,
                    alias = column.name,
                    real_name = column.real_name,
                )
            })
            .collect();

        let query = format!(
            r#"
                CREATE OR REPLACE FUNCTION {up_trigger}()
                RETURNS TRIGGER AS $$
                BEGIN
                    IF NOT reshape.is_new_schema() THEN
                        DECLARE
                            {declarations}
                            {existing_column} public.{table}.{existing_column_real}%TYPE := NEW.{existing_column_real};
                        BEGIN
                            NEW.{temp_column} = {up};
                        END;
                    END IF;
                    RETURN NEW;
                END
                $$ language 'plpgsql';

                DROP TRIGGER IF EXISTS "{up_trigger}" ON "{table}";
                CREATE TRIGGER "{up_trigger}" BEFORE INSERT OR UPDATE ON "{table}" FOR EACH ROW EXECUTE PROCEDURE {up_trigger}();

                CREATE OR REPLACE FUNCTION {down_trigger}()
                RETURNS TRIGGER AS $$
                BEGIN
                    IF reshape.is_new_schema() THEN
                        DECLARE
                            {declarations}
                            {existing_column} public.{table}.{temp_column}%TYPE := NEW.{temp_column};
                        BEGIN
                            NEW.{existing_column_real} = {down};
                        END;
                    END IF;
                    RETURN NEW;
                END
                $$ language 'plpgsql';

                DROP TRIGGER IF EXISTS "{down_trigger}" ON "{table}";
                CREATE TRIGGER "{down_trigger}" BEFORE INSERT OR UPDATE ON "{table}" FOR EACH ROW EXECUTE PROCEDURE {down_trigger}();
                "#,
            existing_column = &self.column,
            existing_column_real = column.real_name,
            temp_column = self.temporary_column_name(ctx),
            up = up,
            down = down,
            table = self.table,
            up_trigger = self.up_trigger_name(ctx),
            down_trigger = self.down_trigger_name(ctx),
            declarations = declarations.join("\n"),
        );
        db.run(&query)
            .context("failed to create up and down triggers")?;

        // Backfill values in batches by touching the previous column
        common::batch_touch_rows(db, &table.real_name, Some(&column.real_name))
            .context("failed to batch update existing rows")?;

        // Duplicate any indices to the temporary column
        let indices = common::get_indices_for_column(db, &table.real_name, &column.real_name)?;
        for index in indices {
            let index_columns: Vec<String> = common::get_index_columns(db, &index.name)?
                .into_iter()
                .map(|idx_column| {
                    // Replace column with temporary column for new index
                    if idx_column == column.real_name {
                        temporary_column_name.to_string()
                    } else {
                        idx_column
                    }
                })
                .collect();
            let temp_index_name = self.temp_index_name(ctx, index.oid);

            let unique_def = if index.unique { "UNIQUE" } else { "" };

            db.query(&format!(
                r#"
                CREATE {unique_def} INDEX CONCURRENTLY IF NOT EXISTS "{new_index_name}" ON "{table}" USING {index_type} ({columns})
                "#,
                new_index_name = temp_index_name,
                table = table.real_name,
                columns = index_columns.join(", "),
                index_type = index.index_type,
            ))
            .context("failed to create temporary index")?;
        }

        // Add a temporary NOT NULL constraint if the column shouldn't be nullable.
        // This constraint is set as NOT VALID so it doesn't apply to existing rows and
        // the existing rows don't need to be scanned under an exclusive lock.
        // Thanks to this, we can set the full column as NOT NULL later with minimal locking.
        if !self.changes.nullable.unwrap_or(column.nullable) {
            let query = format!(
                r#"
                ALTER TABLE "{table}"
                ADD CONSTRAINT "{constraint_name}"
                CHECK ("{column}" IS NOT NULL) NOT VALID
                "#,
                table = self.table,
                constraint_name = self.not_null_constraint_name(ctx),
                column = self.temporary_column_name(ctx),
            );
            db.run(&query)
                .context("failed to add NOT NULL constraint")?;
        }

        Ok(())
    }

    fn complete<'a>(
        &self,
        ctx: &MigrationContext,
        db: &'a mut dyn Conn,
    ) -> anyhow::Result<Option<Transaction<'a>>> {
        if self.can_short_circuit() {
            if let Some(new_name) = &self.changes.name {
                let query = format!(
                    r#"
			        ALTER TABLE "{table}"
			        RENAME COLUMN "{existing_name}" TO "{new_name}"
			        "#,
                    table = self.table,
                    existing_name = self.column,
                    new_name = new_name,
                );
                db.run(&query).context("failed to rename column")?;
            }
            return Ok(None);
        }

        // Update column to be NOT NULL if necessary
        let has_not_null_constraint = !db
            .query_with_params(
                "
                SELECT constraint_name
                FROM information_schema.constraint_column_usage
                WHERE constraint_name = $1
                ",
                &[&self.not_null_constraint_name(ctx)],
            )
            .context("failed to get any NOT NULL constraint")?
            .is_empty();
        if has_not_null_constraint {
            // Validate the temporary constraint (should always be valid).
            // This performs a sequential scan but does not take an exclusive lock.
            let query = format!(
                r#"
                ALTER TABLE "{table}"
                VALIDATE CONSTRAINT "{constraint_name}"
                "#,
                table = self.table,
                constraint_name = self.not_null_constraint_name(ctx),
            );
            db.run(&query)
                .context("failed to validate NOT NULL constraint")?;

            // Update the column to be NOT NULL.
            // This requires an exclusive lock but since PG 12 it can check
            // the existing constraint for correctness which makes the lock short-lived.
            // Source: https://dba.stackexchange.com/a/268128
            let query = format!(
                r#"
                ALTER TABLE "{table}"
                ALTER COLUMN "{column}" SET NOT NULL
                "#,
                table = self.table,
                column = self.temporary_column_name(ctx),
            );
            db.run(&query).context("failed to set column as NOT NULL")?;

            // Drop the temporary constraint
            let query = format!(
                r#"
                ALTER TABLE "{table}"
                DROP CONSTRAINT "{constraint_name}"
                "#,
                table = self.table,
                constraint_name = self.not_null_constraint_name(ctx),
            );
            db.run(&query)
                .context("failed to drop NOT NULL constraint")?;
        }

        // Replace old indices with the new temporary ones created for the temporary column
        let indices = common::get_indices_for_column(db, &self.table, &self.column)?;
        for current_index in indices {
            // To keep the index handling idempotent, we need to do the following:
            // 1. Add a prefix to the existing index
            // 2. Rename temporary index to its final name
            // 3. Drop existing index concurrently

            // Add prefix (if not already added) to existing index
            let prefix = "__reshape_old";
            let target_index_name = current_index.name.trim_start_matches(prefix);
            let old_index_name = format!("{}_{}", prefix, target_index_name);
            db.query(&format!(
                r#"
                ALTER INDEX IF EXISTS "{current_name}" RENAME TO "{new_name}"
                "#,
                current_name = target_index_name,
                new_name = old_index_name,
            ))
            .context("failed to rename old index")?;

            // Rename temporary index to real name
            let temp_index_name = self.temp_index_name(ctx, current_index.oid);
            db.query(&format!(
                r#"
                ALTER INDEX IF EXISTS "{temp_index_name}" RENAME TO "{target_index_name}"
                "#,
                temp_index_name = temp_index_name,
                target_index_name = target_index_name,
            ))
            .context("failed to rename temporary index")?;

            // Drop old index concurrently
            db.query(&format!(
                r#"
                DROP INDEX CONCURRENTLY IF EXISTS "{old_index_name}"
                "#,
                old_index_name = old_index_name,
            ))
            .context("failed to drop old index")?;
        }

        // Remove old column
        let query = format!(
            r#"
            ALTER TABLE "{table}" DROP COLUMN IF EXISTS "{column}" CASCADE
			"#,
            table = self.table,
            column = self.column,
        );
        db.run(&query).context("failed to drop old column")?;

        // Rename temporary column
        let column_name = self.changes.name.as_deref().unwrap_or(&self.column);
        let query = format!(
            r#"
            ALTER TABLE "{table}" RENAME COLUMN "{temp_column}" TO "{name}"
			"#,
            table = self.table,
            temp_column = self.temporary_column_name(ctx),
            name = column_name,
        );
        db.run(&query)
            .context("failed to rename temporary column")?;

        // Remove triggers and procedures
        let query = format!(
            r#"
            DROP TRIGGER IF EXISTS "{up_trigger}" ON "{table}";
            DROP FUNCTION IF EXISTS "{up_trigger}";

            DROP TRIGGER IF EXISTS "{down_trigger}" ON "{table}";
            DROP FUNCTION IF EXISTS "{down_trigger}";
            "#,
            table = self.table,
            up_trigger = self.up_trigger_name(ctx),
            down_trigger = self.down_trigger_name(ctx),
        );
        db.run(&query)
            .context("failed to drop up and down triggers")?;

        Ok(None)
    }

    fn update_schema(&self, ctx: &MigrationContext, schema: &mut Schema) {
        // If we are only changing the name of a column, we haven't created a temporary column
        // Instead, we rename the schema column but point it to the old column
        if self.can_short_circuit() {
            if let Some(new_name) = &self.changes.name {
                schema.change_table(&self.table, |table_changes| {
                    table_changes.change_column(&self.column, |column_changes| {
                        column_changes.set_name(new_name);
                    });
                });
            }

            return;
        }

        schema.change_table(&self.table, |table_changes| {
            table_changes.change_column(&self.column, |column_changes| {
                column_changes.set_column(&self.temporary_column_name(ctx));
            });
        });
    }

    fn abort(&self, ctx: &MigrationContext, db: &mut dyn Conn) -> anyhow::Result<()> {
        // Safely remove any indices created for the temporary column
        let temp_column_name = self.temporary_column_name(ctx);
        let indices = common::get_indices_for_column(db, &self.table, &temp_column_name)?;
        for index in indices {
            let temp_index_name = self.temp_index_name(ctx, index.oid);
            db.query(&format!(
                r#"
                DROP INDEX CONCURRENTLY IF EXISTS "{index_name}"
                "#,
                index_name = temp_index_name,
            ))?;
        }

        // Drop temporary column
        let query = format!(
            r#"
			ALTER TABLE "{table}"
            DROP COLUMN IF EXISTS "{temp_column}";
			"#,
            table = self.table,
            temp_column = self.temporary_column_name(ctx),
        );
        db.run(&query).context("failed to drop temporary column")?;

        // Remove triggers and procedures
        let query = format!(
            r#"
            DROP TRIGGER IF EXISTS "{up_trigger}" ON "{table}";
            DROP FUNCTION IF EXISTS "{up_trigger}";

            DROP TRIGGER IF EXISTS "{down_trigger}" ON "{table}";
            DROP FUNCTION IF EXISTS "{down_trigger}";
            "#,
            table = self.table,
            up_trigger = self.up_trigger_name(ctx),
            down_trigger = self.down_trigger_name(ctx),
        );
        db.run(&query)
            .context("failed to drop up and down triggers")?;

        Ok(())
    }
}

impl AlterColumn {
    fn temporary_column_name(&self, ctx: &MigrationContext) -> String {
        format!("{}_new_{}", ctx.prefix(), self.column)
    }

    fn up_trigger_name(&self, ctx: &MigrationContext) -> String {
        format!("{}_alter_column_up_trigger", ctx.prefix())
    }

    fn down_trigger_name(&self, ctx: &MigrationContext) -> String {
        format!("{}_alter_column_down_trigger", ctx.prefix_inverse())
    }

    fn not_null_constraint_name(&self, ctx: &MigrationContext) -> String {
        format!("{}_alter_column_temporary", ctx.prefix())
    }

    fn temp_index_name(&self, ctx: &MigrationContext, index_oid: u32) -> String {
        format!("{}_alter_column_temp_index_{}", ctx.prefix(), index_oid)
    }

    fn can_short_circuit(&self) -> bool {
        self.changes.name.is_some()
            && self.changes.data_type.is_none()
            && self.changes.nullable.is_none()
            && self.changes.default.is_none()
    }
}



================================================
FILE: src/migrations/common.rs
================================================
use anyhow::anyhow;
use postgres::types::{FromSql, ToSql};
use serde::{Deserialize, Serialize};

use crate::db::Conn;

#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct Column {
    pub name: String,
    #[serde(rename = "type")]
    pub data_type: String,
    #[serde(default = "nullable_default")]
    pub nullable: bool,
    pub default: Option<String>,
    pub generated: Option<String>,
}

fn nullable_default() -> bool {
    true
}

#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct ForeignKey {
    pub columns: Vec<String>,
    pub referenced_table: String,
    pub referenced_columns: Vec<String>,
}

#[derive(Debug)]
struct PostgresRawValue {
    bytes: Vec<u8>,
}

impl<'a> FromSql<'a> for PostgresRawValue {
    fn from_sql(
        _ty: &postgres::types::Type,
        raw: &'a [u8],
    ) -> Result<Self, Box<dyn std::error::Error + Sync + Send>> {
        Ok(PostgresRawValue {
            bytes: raw.to_vec(),
        })
    }

    fn accepts(_ty: &postgres::types::Type) -> bool {
        true
    }
}

impl ToSql for PostgresRawValue {
    fn to_sql(
        &self,
        _ty: &postgres::types::Type,
        out: &mut postgres::types::private::BytesMut,
    ) -> Result<postgres::types::IsNull, Box<dyn std::error::Error + Sync + Send>>
    where
        Self: Sized,
    {
        out.extend_from_slice(&self.bytes);
        Ok(postgres::types::IsNull::No)
    }

    fn accepts(_ty: &postgres::types::Type) -> bool
    where
        Self: Sized,
    {
        true
    }

    postgres::types::to_sql_checked!();
}

pub fn batch_touch_rows(
    db: &mut dyn Conn,
    table: &str,
    column: Option<&str>,
) -> anyhow::Result<()> {
    const BATCH_SIZE: u16 = 1000;

    let mut cursor: Option<PostgresRawValue> = None;

    loop {
        let mut params: Vec<&(dyn ToSql + Sync)> = Vec::new();

        let primary_key = get_primary_key_columns_for_table(db, table)?;

        // If no column to touch is passed, we default to the first primary key column (just to make some "update")
        let touched_column = match column {
            Some(column) => column,
            None => primary_key.first().unwrap(),
        };

        let primary_key_columns = primary_key.join(", ");

        let primary_key_where = primary_key
            .iter()
            .map(|column| {
                format!(
                    r#"
                    "{table}"."{column}" = rows."{column}"
                    "#,
                    table = table,
                    column = column,
                )
            })
            .collect::<Vec<String>>()
            .join(" AND ");

        let returning_columns = primary_key
            .iter()
            .map(|column| format!("rows.\"{}\"", column))
            .collect::<Vec<String>>()
            .join(", ");

        let cursor_where = if let Some(cursor) = &cursor {
            params.push(cursor);

            format!(
                "WHERE ({primary_key_columns}) > $1",
                primary_key_columns = primary_key_columns
            )
        } else {
            "".to_string()
        };

        let query = format!(
            r#"
            WITH rows AS (
                SELECT {primary_key_columns}
                FROM public."{table}"
                {cursor_where}
                ORDER BY {primary_key_columns}
                LIMIT {batch_size}
            ), update AS (
                UPDATE public."{table}" "{table}"
                SET "{touched_column}" = "{table}"."{touched_column}"
                FROM rows
                WHERE {primary_key_where}
                RETURNING {returning_columns}
            )
            SELECT LAST_VALUE(({primary_key_columns})) OVER () AS last_value
            FROM update
            LIMIT 1
            "#,
            batch_size = BATCH_SIZE,
        );
        let last_value = db
            .query_with_params(&query, &params)?
            .first()
            .and_then(|row| row.get("last_value"));

        if last_value.is_none() {
            break;
        }

        cursor = last_value
    }

    Ok(())
}

fn get_primary_key_columns_for_table(
    db: &mut dyn Conn,
    table: &str,
) -> anyhow::Result<Vec<String>> {
    // Query from https://wiki.postgresql.org/wiki/Retrieve_primary_key_columns
    let primary_key_columns: Vec<String> = db
        .query(&format!(
            "
            SELECT a.attname AS column_name
            FROM   pg_index i
            JOIN   pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey)
            WHERE  i.indrelid = '{table}'::regclass
            AND    i.indisprimary;
            ",
            table = table
        ))?
        .iter()
        .map(|row| row.get("column_name"))
        .collect();

    Ok(primary_key_columns)
}

pub struct Index {
    pub name: String,
    pub oid: u32,
    pub unique: bool,
    pub index_type: String,
}

pub fn get_indices_for_column(
    db: &mut dyn Conn,
    table: &str,
    column: &str,
) -> anyhow::Result<Vec<Index>> {
    let indices = db
        .query(&format!(
            "
            SELECT
                i.relname AS name,
                i.oid AS oid,
                ix.indisunique AS unique,
                am.amname AS type
            FROM pg_index ix
            JOIN pg_class t ON t.oid = ix.indrelid
            JOIN pg_class i ON i.oid = ix.indexrelid
            JOIN pg_am am ON i.relam = am.oid
            JOIN pg_attribute a ON
                a.attrelid = t.oid AND
                a.attnum = ANY(ix.indkey)
            WHERE
                t.relname = '{table}' AND
                a.attname = '{column}'
            ",
            table = table,
            column = column,
        ))?
        .iter()
        .map(|row| Index {
            name: row.get("name"),
            oid: row.get("oid"),
            unique: row.get("unique"),
            index_type: row.get("type"),
        })
        .collect();

    Ok(indices)
}

pub fn get_index_columns(db: &mut dyn Conn, index_name: &str) -> anyhow::Result<Vec<String>> {
    // Get all columns which are part of the index in order
    let (table_oid, column_nums) = db
        .query(&format!(
            "
            SELECT t.oid AS table_oid, ix.indkey::INTEGER[] AS columns
            FROM pg_index ix
            JOIN pg_class t ON t.oid = ix.indrelid
            JOIN pg_class i ON i.oid = ix.indexrelid
            WHERE
	            i.relname = '{index_name}'
            ",
            index_name = index_name,
        ))?
        .first()
        .map(|row| {
            (
                row.get::<'_, _, u32>("table_oid"),
                row.get::<'_, _, Vec<i32>>("columns"),
            )
        })
        .ok_or_else(|| anyhow!("failed to get columns for index"))?;

    // Get the name of each of the columns, still in order
    column_nums
        .iter()
        .map(|column_num| -> anyhow::Result<String> {
            let name: String = db
                .query(&format!(
                    "
                    SELECT attname AS name
                    FROM pg_attribute
                    WHERE attrelid = {table_oid}
                        AND attnum = {column_num};
                    ",
                    table_oid = table_oid,
                    column_num = column_num,
                ))?
                .first()
                .map(|row| row.get("name"))
                .ok_or_else(|| anyhow!("expected to find column"))?;

            Ok(name)
        })
        .collect::<anyhow::Result<Vec<String>>>()
}



================================================
FILE: src/migrations/create_enum.rs
================================================
use super::{Action, MigrationContext};
use crate::{
    db::{Conn, Transaction},
    schema::Schema,
};
use anyhow::Context;
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize, Debug)]
pub struct CreateEnum {
    pub name: String,
    pub values: Vec<String>,
}

#[typetag::serde(name = "create_enum")]
impl Action for CreateEnum {
    fn describe(&self) -> String {
        format!("Creating enum \"{}\"", self.name)
    }

    fn run(
        &self,
        _ctx: &MigrationContext,
        db: &mut dyn Conn,
        _schema: &Schema,
    ) -> anyhow::Result<()> {
        // Check if enum already exists. CREATE TYPE doesn't have
        // a IF NOT EXISTS option so we have to do it manually.
        let enum_exists = !db
            .query(&format!(
                "
                SELECT typname
                FROM pg_catalog.pg_type
                WHERE typcategory = 'E'
                AND typname = '{name}'
                ",
                name = self.name,
            ))?
            .is_empty();
        if enum_exists {
            return Ok(());
        }

        let values_def: Vec<String> = self
            .values
            .iter()
            .map(|value| format!("'{}'", value))
            .collect();

        db.run(&format!(
            r#"
            CREATE TYPE "{name}" AS ENUM ({values})
            "#,
            name = self.name,
            values = values_def.join(", "),
        ))
        .context("failed to create enum")?;

        Ok(())
    }

    fn complete<'a>(
        &self,
        _ctx: &MigrationContext,
        _db: &'a mut dyn Conn,
    ) -> anyhow::Result<Option<Transaction<'a>>> {
        Ok(None)
    }

    fn update_schema(&self, _ctx: &MigrationContext, _schema: &mut Schema) {}

    fn abort(&self, _ctx: &MigrationContext, db: &mut dyn Conn) -> anyhow::Result<()> {
        db.run(&format!(
            r#"
            DROP TYPE IF EXISTS {name}
            "#,
            name = self.name,
        ))
        .context("failed to drop enum")?;

        Ok(())
    }
}



================================================
FILE: src/migrations/create_table.rs
================================================
use std::collections::HashMap;

use super::{common::ForeignKey, Action, Column, MigrationContext};
use crate::{
    db::{Conn, Transaction},
    migrations::common,
    schema::Schema,
};
use anyhow::Context;
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize, Debug)]
pub struct CreateTable {
    pub name: String,
    pub columns: Vec<Column>,
    pub primary_key: Vec<String>,

    #[serde(default)]
    pub foreign_keys: Vec<ForeignKey>,

    pub up: Option<Transformation>,
}

#[derive(Serialize, Deserialize, Debug)]
pub struct Transformation {
    table: String,
    values: HashMap<String, String>,
    upsert_constraint: Option<String>,
}

impl CreateTable {
    fn trigger_name(&self, ctx: &MigrationContext) -> String {
        format!("{}_create_table_{}", ctx.prefix(), self.name)
    }
}

#[typetag::serde(name = "create_table")]
impl Action for CreateTable {
    fn describe(&self) -> String {
        format!("Creating table \"{}\"", self.name)
    }

    fn run(
        &self,
        ctx: &MigrationContext,
        db: &mut dyn Conn,
        schema: &Schema,
    ) -> anyhow::Result<()> {
        let mut definition_rows: Vec<String> = self
            .columns
            .iter()
            .map(|column| {
                let mut parts = vec![format!("\"{}\"", column.name), column.data_type.to_string()];

                if let Some(default) = &column.default {
                    parts.push("DEFAULT".to_string());
                    parts.push(default.to_string());
                }

                if !column.nullable {
                    parts.push("NOT NULL".to_string());
                }

                if let Some(generated) = &column.generated {
                    parts.push("GENERATED".to_string());
                    parts.push(generated.to_string());
                }

                parts.join(" ")
            })
            .collect();

        let primary_key_columns = self
            .primary_key
            .iter()
            // Add quotes around all column names
            .map(|col| format!("\"{}\"", col))
            .collect::<Vec<String>>()
            .join(", ");
        definition_rows.push(format!("PRIMARY KEY ({})", primary_key_columns));

        for foreign_key in &self.foreign_keys {
            // Add quotes around all column names
            let columns: Vec<String> = foreign_key
                .columns
                .iter()
                .map(|col| format!("\"{}\"", col))
                .collect();

            let referenced_table = schema.get_table(db, &foreign_key.referenced_table)?;
            let referenced_columns: Vec<String> = referenced_table
                .real_column_names(&foreign_key.referenced_columns)
                .map(|col| format!("\"{}\"", col))
                .collect();

            definition_rows.push(format!(
                r#"
                FOREIGN KEY ({columns}) REFERENCES "{table}" ({referenced_columns})
                "#,
                columns = columns.join(", "),
                table = referenced_table.real_name,
                referenced_columns = referenced_columns.join(", "),
            ));
        }

        let query = &format!(
            r#"
            CREATE TABLE "{name}" (
                {definition}
            )
            "#,
            name = self.name,
            definition = definition_rows.join(",\n"),
        );
        db.run(query).context("failed to create table")?;

        if let Some(Transformation {
            table: from_table,
            values,
            upsert_constraint,
        }) = &self.up
        {
            let from_table = schema.get_table(db, &from_table)?;

            let declarations: Vec<String> = from_table
                .columns
                .iter()
                .map(|column| {
                    format!(
                        "{alias} public.{table}.{real_name}%TYPE := NEW.{real_name};",
                        table = from_table.real_name,
                        alias = column.name,
                        real_name = column.real_name,
                    )
                })
                .collect();

            let (insert_columns, insert_values): (Vec<&str>, Vec<&str>) = values
                .iter()
                .map(|(k, v)| -> (&str, &str) { (k, v) }) // Force &String to &str
                .unzip();

            let update_set: Vec<String> = values
                .iter()
                .map(|(field, value)| format!("\"{field}\" = {value}"))
                .collect();

            // Constraint to check for conflicts. Defaults to the primary key constraint.
            let conflict_constraint_name = match upsert_constraint {
                Some(custom_constraint) => custom_constraint.clone(),
                _ => format!("{table}_pkey", table = self.name),
            };

            // Add triggers to fill in values as they are inserted/updated
            let query = format!(
                r#"
                    CREATE OR REPLACE FUNCTION {trigger_name}()
                    RETURNS TRIGGER AS $$
                    #variable_conflict use_variable
                    BEGIN
                        IF NOT reshape.is_new_schema() THEN
                            DECLARE
                                {declarations}
                            BEGIN
                                INSERT INTO public."{changed_table_real}" ({columns})
                                VALUES ({values})
                                ON CONFLICT ON CONSTRAINT "{conflict_constraint_name}"
                                DO UPDATE SET
                                    {updates};
                            END;
                        END IF;
                        RETURN NEW;
                    END
                    $$ language 'plpgsql';

                    DROP TRIGGER IF EXISTS "{trigger_name}" ON "{from_table_real}";
                    CREATE TRIGGER "{trigger_name}" BEFORE UPDATE OR INSERT ON "{from_table_real}" FOR EACH ROW EXECUTE PROCEDURE {trigger_name}();
                    "#,
                changed_table_real = self.name,
                from_table_real = from_table.real_name,
                trigger_name = self.trigger_name(ctx),
                declarations = declarations.join("\n"),
                columns = insert_columns.join(", "),
                values = insert_values.join(", "),
                updates = update_set.join(",\n"),
            );
            db.run(&query).context("failed to create up trigger")?;

            // Backfill values in batches by touching the from table
            common::batch_touch_rows(db, &from_table.real_name, None)
                .context("failed to batch update existing rows")?;
        }

        Ok(())
    }

    fn complete<'a>(
        &self,
        ctx: &MigrationContext,
        db: &'a mut dyn Conn,
    ) -> anyhow::Result<Option<Transaction<'a>>> {
        // Remove triggers and procedures
        let query = format!(
            r#"
            DROP FUNCTION IF EXISTS "{trigger_name}" CASCADE;
            "#,
            trigger_name = self.trigger_name(ctx),
        );
        db.run(&query).context("failed to drop up trigger")?;

        Ok(None)
    }

    fn update_schema(&self, _ctx: &MigrationContext, _schema: &mut Schema) {}

    fn abort(&self, ctx: &MigrationContext, db: &mut dyn Conn) -> anyhow::Result<()> {
        // Remove triggers and procedures
        let query = format!(
            r#"
            DROP FUNCTION IF EXISTS "{trigger_name}" CASCADE;
            "#,
            trigger_name = self.trigger_name(ctx),
        );
        db.run(&query).context("failed to drop up trigger")?;

        db.run(&format!(
            r#"
            DROP TABLE IF EXISTS "{name}"
            "#,
            name = self.name,
        ))
        .context("failed to drop table")?;

        Ok(())
    }
}



================================================
FILE: src/migrations/custom.rs
================================================
use super::{Action, MigrationContext};
use crate::{
    db::{Conn, Transaction},
    schema::Schema,
};
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize, Debug)]
pub struct Custom {
    #[serde(default)]
    pub start: Option<String>,

    #[serde(default)]
    pub complete: Option<String>,

    #[serde(default)]
    pub abort: Option<String>,
}

#[typetag::serde(name = "custom")]
impl Action for Custom {
    fn describe(&self) -> String {
        "Running custom migration".to_string()
    }

    fn run(
        &self,
        _ctx: &MigrationContext,
        db: &mut dyn Conn,
        _schema: &Schema,
    ) -> anyhow::Result<()> {
        if let Some(start_query) = &self.start {
            println!("Running query: {}", start_query);
            db.run(start_query)?;
        }

        Ok(())
    }

    fn complete<'a>(
        &self,
        _ctx: &MigrationContext,
        db: &'a mut dyn Conn,
    ) -> anyhow::Result<Option<Transaction<'a>>> {
        if let Some(complete_query) = &self.complete {
            db.run(complete_query)?;
        }

        Ok(None)
    }

    fn update_schema(&self, _ctx: &MigrationContext, _schema: &mut Schema) {}

    fn abort(&self, _ctx: &MigrationContext, db: &mut dyn Conn) -> anyhow::Result<()> {
        if let Some(abort_query) = &self.abort {
            db.run(abort_query)?;
        }

        Ok(())
    }
}



================================================
FILE: src/migrations/mod.rs
================================================
use crate::{
    db::{Conn, Transaction},
    schema::Schema,
};
use core::fmt::Debug;
use serde::{Deserialize, Serialize};

// Re-export migration types
mod common;
pub use common::Column;

mod create_table;
pub use create_table::CreateTable;

mod alter_column;
pub use alter_column::{AlterColumn, ColumnChanges};

mod add_column;
pub use add_column::AddColumn;

mod remove_column;
pub use remove_column::RemoveColumn;

mod add_index;
pub use add_index::{AddIndex, Index};

mod remove_index;
pub use remove_index::RemoveIndex;

mod remove_table;
pub use remove_table::RemoveTable;

mod rename_table;
pub use rename_table::RenameTable;

mod create_enum;
pub use create_enum::CreateEnum;

mod remove_enum;
pub use remove_enum::RemoveEnum;

mod custom;
pub use custom::Custom;

mod add_foreign_key;
pub use add_foreign_key::AddForeignKey;

mod remove_foreign_key;
pub use remove_foreign_key::RemoveForeignKey;

#[derive(Serialize, Deserialize, Debug)]
pub struct Migration {
    pub name: String,
    pub description: Option<String>,
    pub actions: Vec<Box<dyn Action>>,
}

impl Migration {
    pub fn new(name: impl Into<String>, description: Option<String>) -> Migration {
        Migration {
            name: name.into(),
            description,
            actions: vec![],
        }
    }

    pub fn with_action(mut self, action: impl Action + 'static) -> Self {
        self.actions.push(Box::new(action));
        self
    }
}

impl PartialEq for Migration {
    fn eq(&self, other: &Self) -> bool {
        self.name == other.name
    }
}

impl Eq for Migration {}

impl Clone for Migration {
    fn clone(&self) -> Self {
        let serialized = serde_json::to_string(self).unwrap();
        serde_json::from_str(&serialized).unwrap()
    }
}

pub struct MigrationContext {
    migration_index: usize,
    action_index: usize,
    existing_schema_name: Option<String>,
}

impl MigrationContext {
    pub fn new(
        migration_index: usize,
        action_index: usize,
        existing_schema_name: Option<String>,
    ) -> Self {
        MigrationContext {
            migration_index,
            action_index,
            existing_schema_name,
        }
    }

    fn prefix(&self) -> String {
        format!(
            "__reshape_{:0>4}_{:0>4}",
            self.migration_index, self.action_index
        )
    }

    fn prefix_inverse(&self) -> String {
        format!(
            "__reshape_{:0>4}_{:0>4}",
            1000 - self.migration_index,
            1000 - self.action_index
        )
    }
}

#[typetag::serde(tag = "type")]
pub trait Action: Debug {
    fn describe(&self) -> String;
    fn run(&self, ctx: &MigrationContext, db: &mut dyn Conn, schema: &Schema)
        -> anyhow::Result<()>;
    fn complete<'a>(
        &self,
        ctx: &MigrationContext,
        db: &'a mut dyn Conn,
    ) -> anyhow::Result<Option<Transaction<'a>>>;
    fn update_schema(&self, ctx: &MigrationContext, schema: &mut Schema);
    fn abort(&self, ctx: &MigrationContext, db: &mut dyn Conn) -> anyhow::Result<()>;
}



================================================
FILE: src/migrations/remove_column.rs
================================================
use super::{common, Action, MigrationContext};
use crate::{
    db::{Conn, Transaction},
    schema::Schema,
};
use anyhow::{anyhow, bail, Context};
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize, Debug)]
pub struct RemoveColumn {
    pub table: String,
    pub column: String,
    pub down: Option<Transformation>,
}

#[derive(Serialize, Deserialize, Debug)]
#[serde(untagged)]
pub enum Transformation {
    Simple(String),
    Update {
        table: String,
        value: String,
        r#where: String,
    },
}

impl RemoveColumn {
    fn trigger_name(&self, ctx: &MigrationContext) -> String {
        format!(
            "{}_remove_column_{}_{}",
            ctx.prefix(),
            self.table,
            self.column
        )
    }

    fn reverse_trigger_name(&self, ctx: &MigrationContext) -> String {
        format!(
            "{}_remove_column_{}_{}_rev",
            ctx.prefix(),
            self.table,
            self.column
        )
    }

    fn not_null_constraint_trigger_name(&self, ctx: &MigrationContext) -> String {
        format!(
            "{}_remove_column_{}_{}_nn",
            ctx.prefix(),
            self.table,
            self.column
        )
    }

    fn not_null_constraint_name(&self, ctx: &MigrationContext) -> String {
        format!(
            "{}_add_column_not_null_{}_{}",
            ctx.prefix(),
            self.table,
            self.column
        )
    }
}

#[typetag::serde(name = "remove_column")]
impl Action for RemoveColumn {
    fn describe(&self) -> String {
        format!(
            "Removing column \"{}\" from \"{}\"",
            self.column, self.table
        )
    }

    fn run(
        &self,
        ctx: &MigrationContext,
        db: &mut dyn Conn,
        schema: &Schema,
    ) -> anyhow::Result<()> {
        let table = schema.get_table(db, &self.table)?;
        let column = table
            .get_column(&self.column)
            .ok_or_else(|| anyhow!("no such column {} exists", self.column))?;

        // Add down trigger
        if let Some(down) = &self.down {
            let declarations: Vec<String> = table
                .columns
                .iter()
                .map(|column| {
                    format!(
                        "{alias} public.{table}.{real_name}%TYPE := NEW.{real_name};",
                        table = table.real_name,
                        alias = column.name,
                        real_name = column.real_name,
                    )
                })
                .collect();

            if let Transformation::Simple(down) = down {
                let query = format!(
                    r#"
                    CREATE OR REPLACE FUNCTION {trigger_name}()
                    RETURNS TRIGGER AS $$
                    BEGIN
                        IF reshape.is_new_schema() THEN
                            DECLARE
                                {declarations}
                            BEGIN
                                NEW.{column_name} = {down};
                            END;
                        END IF;
                        RETURN NEW;
                    END
                    $$ language 'plpgsql';

                    DROP TRIGGER IF EXISTS "{trigger_name}" ON "{table}";
                    CREATE TRIGGER "{trigger_name}" BEFORE UPDATE OR INSERT ON "{table}" FOR EACH ROW EXECUTE PROCEDURE {trigger_name}();
                    "#,
                    column_name = self.column,
                    trigger_name = self.trigger_name(ctx),
                    down = down,
                    table = self.table,
                    declarations = declarations.join("\n"),
                );
                db.run(&query).context("failed to create down trigger")?;
            }

            if let Transformation::Update {
                table: from_table,
                value,
                r#where,
            } = down
            {
                let existing_schema_name = match &ctx.existing_schema_name {
                    Some(name) => name,
                    None => bail!("can't use update without previous migration"),
                };

                let from_table = schema.get_table(db, &from_table)?;

                let maybe_null_check = if !column.nullable {
                    // Replace NOT NULL constraint with a constraint trigger that only triggers on the old schema.
                    // We will add a null check to the down function on the new schema below as well to cover both cases.
                    // As we are using a complex down function, we must remove the NOT NULL check for the new schema.
                    // NOT NULL is not checked at the end of a transaction, but immediately upon update.
                    let query = format!(
                        r#"
                        CREATE OR REPLACE FUNCTION {trigger_name}()
                        RETURNS TRIGGER AS $$
                        BEGIN
                            IF NOT reshape.is_new_schema() THEN
                                IF NEW.{column} IS NULL THEN
                                    RAISE EXCEPTION '{column} can not be null';
                                END IF;
                            END IF;
                            RETURN NEW;
                        END
                        $$ language 'plpgsql';

                        DROP TRIGGER IF EXISTS "{trigger_name}" ON "{table}";

                        CREATE CONSTRAINT TRIGGER "{trigger_name}"
                            AFTER INSERT OR UPDATE
                            ON "{table}"
                            FOR EACH ROW
                            EXECUTE PROCEDURE {trigger_name}();
                        "#,
                        table = self.table,
                        trigger_name = self.not_null_constraint_trigger_name(ctx),
                        column = self.column,
                    );
                    db.run(&query)
                        .context("failed to create null constraint trigger")?;

                    db.run(&format!(
                        r#"
                        ALTER TABLE {table}
                        ALTER COLUMN {column}
                        DROP NOT NULL
                        "#,
                        table = self.table,
                        column = self.column
                    ))
                    .context("failed to remove column not null constraint")?;

                    format!(
                        r#"
                        IF {value} IS NULL THEN
                            RAISE EXCEPTION '{column_name} can not be null';
                        END IF;
                        "#,
                        column_name = self.column,
                    )
                } else {
                    "".to_string()
                };

                let into_variables = from_table
                    .columns
                    .iter()
                    .map(|column| {
                        format!(
                            "NEW.{real_name} AS {alias}",
                            alias = column.name,
                            real_name = column.real_name,
                        )
                    })
                    .collect::<Vec<String>>()
                    .join(", ");

                let query = format!(
                    r#"
                    CREATE OR REPLACE FUNCTION {trigger_name}()
                    RETURNS TRIGGER AS $$
                    #variable_conflict use_variable
                    BEGIN
                        IF reshape.is_new_schema() THEN
                            DECLARE
                                {from_table} record;
                            BEGIN
                                SELECT {into_variables}
                                INTO {from_table};

                                {maybe_null_check}

                                -- Don't trigger reverse trigger when making this update
                                perform set_config('reshape.disable_triggers', 'TRUE', TRUE);

                                UPDATE "migration_{existing_schema_name}"."{changed_table}" "{changed_table}"
                                SET "{column_name}" = {value}
                                WHERE {where};

                                perform set_config('reshape.disable_triggers', '', TRUE);
                            END;
                        END IF;
                        RETURN NEW;
                    END
                    $$ language 'plpgsql';

                    DROP TRIGGER IF EXISTS "{trigger_name}" ON "{from_table_real}";
                    CREATE TRIGGER "{trigger_name}" BEFORE UPDATE OR INSERT ON "{from_table_real}" FOR EACH ROW EXECUTE PROCEDURE {trigger_name}();
                    "#,
                    changed_table = self.table,
                    from_table = from_table.name,
                    from_table_real = from_table.real_name,
                    column_name = self.column,
                    trigger_name = self.trigger_name(ctx),
                );
                db.run(&query).context("failed to create down trigger")?;

                let changed_into_variables = table
                    .columns
                    .iter()
                    .map(|column| {
                        format!(
                            "NEW.{real_name} AS {alias}",
                            alias = column.name,
                            real_name = column.real_name,
                        )
                    })
                    .collect::<Vec<String>>()
                    .join(", ");

                let from_table_columns = from_table
                    .columns
                    .iter()
                    .map(|column| format!("{} as {}", column.real_name, column.name))
                    .collect::<Vec<String>>()
                    .join(", ");

                let query = format!(
                    r#"
                    CREATE OR REPLACE FUNCTION {trigger_name}()
                    RETURNS TRIGGER AS $$
                    #variable_conflict use_variable
                    BEGIN
                        IF reshape.is_new_schema() AND NOT current_setting('reshape.disable_triggers', TRUE) = 'TRUE' THEN
                            DECLARE
                                {changed_table} record;
                                __temp_row record;
                            BEGIN
                                SELECT {changed_into_variables}
                                INTO {changed_table};

                                SELECT *
                                INTO __temp_row
                                FROM (
                                    SELECT {from_table_columns}
                                    FROM public.{from_table_real}
                                ) {from_table}
                                WHERE {where};

                                DECLARE
                                    {from_table} record;
                                BEGIN
                                    {from_table} := __temp_row;
                                    NEW.{column_name_real} = {value};
                                END;
                            END;
                        END IF;
                        RETURN NEW;
                    END
                    $$ language 'plpgsql';

                    DROP TRIGGER IF EXISTS "{trigger_name}" ON "{changed_table_real}";
                    CREATE TRIGGER "{trigger_name}" BEFORE UPDATE OR INSERT ON "{changed_table_real}" FOR EACH ROW EXECUTE PROCEDURE {trigger_name}();
                    "#,
                    changed_table = table.name,
                    changed_table_real = table.real_name,
                    from_table = from_table.name,
                    from_table_real = from_table.real_name,
                    column_name_real = column.real_name,
                    trigger_name = self.reverse_trigger_name(ctx),
                    // declarations = declarations.join("\n"),
                );
                db.run(&query)
                    .context("failed to create reverse down trigger")?;
            }
        }

        Ok(())
    }

    fn complete<'a>(
        &self,
        ctx: &MigrationContext,
        db: &'a mut dyn Conn,
    ) -> anyhow::Result<Option<Transaction<'a>>> {
        let indices = common::get_indices_for_column(db, &self.table, &self.column)
            .context("failed getting column indices")?;

        for index in indices {
            db.run(&format!(
                "
                DROP INDEX CONCURRENTLY IF EXISTS {name}
                ",
                name = index.name,
            ))
            .context("failed to drop index")?;
        }

        // Remove column, function and trigger
        let query = format!(
            r#"
            ALTER TABLE "{table}"
            DROP COLUMN IF EXISTS "{column}";

            DROP FUNCTION IF EXISTS "{trigger_name}" CASCADE;
            DROP FUNCTION IF EXISTS "{reverse_trigger_name}" CASCADE;
            DROP FUNCTION IF EXISTS "{null_trigger_name}" CASCADE;
            "#,
            table = self.table,
            column = self.column,
            trigger_name = self.trigger_name(ctx),
            reverse_trigger_name = self.reverse_trigger_name(ctx),
            null_trigger_name = self.not_null_constraint_trigger_name(ctx),
        );
        db.run(&query)
            .context("failed to drop column and down trigger")?;

        Ok(None)
    }

    fn update_schema(&self, _ctx: &MigrationContext, schema: &mut Schema) {
        schema.change_table(&self.table, |table_changes| {
            table_changes.change_column(&self.column, |column_changes| {
                column_changes.set_removed();
            })
        });
    }

    fn abort(&self, ctx: &MigrationContext, db: &mut dyn Conn) -> anyhow::Result<()> {
        // We might have temporaily removed the NOT NULL check and have to reinstate it
        let has_not_null_function = !db
            .query_with_params(
                "
                SELECT routine_name
                FROM information_schema.routines
                WHERE routine_schema = 'public'
                AND routine_name = $1
                ",
                &[&self.not_null_constraint_trigger_name(ctx)],
            )
            .context("failed to get any NOT NULL function")?
            .is_empty();

        if has_not_null_function {
            // Make column NOT NULL again without taking any long lived locks with a temporary constraint
            let query = format!(
                r#"
                 ALTER TABLE "{table}"
                 ADD CONSTRAINT "{constraint_name}"
                 CHECK ("{column}" IS NOT NULL) NOT VALID
                 "#,
                table = self.table,
                constraint_name = self.not_null_constraint_name(ctx),
                column = self.column,
            );
            db.run(&query)
                .context("failed to add NOT NULL constraint")?;

            let query = format!(
                r#"
                ALTER TABLE "{table}"
                VALIDATE CONSTRAINT "{constraint_name}"
                "#,
                table = self.table,
                constraint_name = self.not_null_constraint_name(ctx),
            );
            db.run(&query)
                .context("failed to validate NOT NULL constraint")?;

            // This ALTER TABLE call will not require any exclusive locks as it can use the validated constraint from above
            db.run(&format!(
                r#"
                ALTER TABLE {table}
                ALTER COLUMN {column}
                SET NOT NULL
                "#,
                table = self.table,
                column = self.column
            ))
            .context("failed to reinstate column NOT NULL")?;

            // Drop the temporary constraint
            let query = format!(
                r#"
                ALTER TABLE "{table}"
                DROP CONSTRAINT "{constraint_name}"
                "#,
                table = self.table,
                constraint_name = self.not_null_constraint_name(ctx),
            );
            db.run(&query)
                .context("failed to drop NOT NULL constraint")?;
        }

        // Remove function and trigger
        db.run(&format!(
            r#"
            DROP FUNCTION IF EXISTS "{trigger_name}" CASCADE;
            DROP FUNCTION IF EXISTS "{reverse_trigger_name}" CASCADE;
            DROP FUNCTION IF EXISTS "{null_trigger_name}" CASCADE;
            "#,
            trigger_name = self.trigger_name(ctx),
            reverse_trigger_name = self.reverse_trigger_name(ctx),
            null_trigger_name = self.not_null_constraint_trigger_name(ctx),
        ))
        .context("failed to drop down trigger")?;

        Ok(())
    }
}



================================================
FILE: src/migrations/remove_enum.rs
================================================
use super::{Action, MigrationContext};
use crate::{
    db::{Conn, Transaction},
    schema::Schema,
};
use anyhow::Context;
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize, Debug)]
pub struct RemoveEnum {
    #[serde(rename = "enum")]
    pub enum_name: String,
}

#[typetag::serde(name = "remove_enum")]
impl Action for RemoveEnum {
    fn describe(&self) -> String {
        format!("Removing enum \"{}\"", self.enum_name)
    }

    fn run(
        &self,
        _ctx: &MigrationContext,
        _db: &mut dyn Conn,
        _schema: &Schema,
    ) -> anyhow::Result<()> {
        Ok(())
    }

    fn complete<'a>(
        &self,
        _ctx: &MigrationContext,
        db: &'a mut dyn Conn,
    ) -> anyhow::Result<Option<Transaction<'a>>> {
        db.run(&format!(
            r#"
            DROP TYPE IF EXISTS {name}
            "#,
            name = self.enum_name,
        ))
        .context("failed to drop enum")?;

        Ok(None)
    }

    fn update_schema(&self, _ctx: &MigrationContext, _schema: &mut Schema) {}

    fn abort(&self, _ctx: &MigrationContext, _db: &mut dyn Conn) -> anyhow::Result<()> {
        Ok(())
    }
}



================================================
FILE: src/migrations/remove_foreign_key.rs
================================================
use super::{Action, MigrationContext};
use crate::{
    db::{Conn, Transaction},
    schema::Schema,
};
use anyhow::{anyhow, Context};
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize, Debug)]
pub struct RemoveForeignKey {
    table: String,
    foreign_key: String,
}

#[typetag::serde(name = "remove_foreign_key")]
impl Action for RemoveForeignKey {
    fn describe(&self) -> String {
        format!(
            "Removing foreign key \"{}\" from table \"{}\"",
            self.foreign_key, self.table
        )
    }

    fn run(
        &self,
        _ctx: &MigrationContext,
        db: &mut dyn Conn,
        schema: &Schema,
    ) -> anyhow::Result<()> {
        // The foreign key is only removed once the migration is completed.
        // Removing it earlier would be hard/undesirable for several reasons:
        // - Postgres doesn't have an easy way to temporarily disable a foreign key check.
        //   If it did, we could disable the FK for the new schema.
        // - Even if we could, it probably wouldn't be a good idea as it would cause temporary
        //   inconsistencies for the old schema which still expects the FK to hold.
        // - For the same reason, we can't remove the FK when the migration is first applied.
        //   If the migration was to be aborted, then the FK would have to be recreated with
        //   the risk that it would no longer be valid.

        // Ensure foreign key exists
        let table = schema.get_table(db, &self.table)?;
        let fk_exists = !db
            .query(&format!(
                r#"
                SELECT constraint_name
                FROM information_schema.table_constraints
                WHERE
                    constraint_type = 'FOREIGN KEY' AND
                    table_name = '{table_name}' AND
                    constraint_name = '{foreign_key}' 
                "#,
                table_name = table.real_name,
                foreign_key = self.foreign_key,
            ))
            .context("failed to check for foreign key")?
            .is_empty();

        if !fk_exists {
            return Err(anyhow!(
                "no foreign key \"{}\" exists on table \"{}\"",
                self.foreign_key,
                self.table
            ));
        }

        Ok(())
    }

    fn complete<'a>(
        &self,
        _ctx: &MigrationContext,
        db: &'a mut dyn Conn,
    ) -> anyhow::Result<Option<Transaction<'a>>> {
        db.run(&format!(
            r#"
            ALTER TABLE {table}
            DROP CONSTRAINT IF EXISTS {foreign_key}
            "#,
            table = self.table,
            foreign_key = self.foreign_key,
        ))
        .context("failed to remove foreign key")?;
        Ok(None)
    }

    fn update_schema(&self, _ctx: &MigrationContext, _schema: &mut Schema) {}

    fn abort(&self, _ctx: &MigrationContext, _db: &mut dyn Conn) -> anyhow::Result<()> {
        Ok(())
    }
}



================================================
FILE: src/migrations/remove_index.rs
================================================
use super::{Action, MigrationContext};
use crate::{
    db::{Conn, Transaction},
    schema::Schema,
};
use anyhow::Context;
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize, Debug)]
pub struct RemoveIndex {
    pub index: String,
}

#[typetag::serde(name = "remove_index")]
impl Action for RemoveIndex {
    fn describe(&self) -> String {
        format!("Removing index \"{}\"", self.index)
    }

    fn run(
        &self,
        _ctx: &MigrationContext,
        _db: &mut dyn Conn,
        _schema: &Schema,
    ) -> anyhow::Result<()> {
        // Do nothing, the index isn't removed until completion
        Ok(())
    }

    fn complete<'a>(
        &self,
        _ctx: &MigrationContext,
        db: &'a mut dyn Conn,
    ) -> anyhow::Result<Option<Transaction<'a>>> {
        db.run(&format!(
            r#"
            DROP INDEX CONCURRENTLY IF EXISTS "{name}"
            "#,
            name = self.index
        ))
        .context("failed to drop index")?;

        Ok(None)
    }

    fn update_schema(&self, _ctx: &MigrationContext, _schema: &mut Schema) {}

    fn abort(&self, _ctx: &MigrationContext, _db: &mut dyn Conn) -> anyhow::Result<()> {
        Ok(())
    }
}



================================================
FILE: src/migrations/remove_table.rs
================================================
use super::{Action, MigrationContext};
use crate::{
    db::{Conn, Transaction},
    schema::Schema,
};
use anyhow::Context;
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize, Debug)]
pub struct RemoveTable {
    pub table: String,
}

#[typetag::serde(name = "remove_table")]
impl Action for RemoveTable {
    fn describe(&self) -> String {
        format!("Removing table \"{}\"", self.table)
    }

    fn run(
        &self,
        _ctx: &MigrationContext,
        _db: &mut dyn Conn,
        _schema: &Schema,
    ) -> anyhow::Result<()> {
        Ok(())
    }

    fn complete<'a>(
        &self,
        _ctx: &MigrationContext,
        db: &'a mut dyn Conn,
    ) -> anyhow::Result<Option<Transaction<'a>>> {
        // Remove table
        let query = format!(
            r#"
            DROP TABLE IF EXISTS "{table}";
            "#,
            table = self.table,
        );
        db.run(&query).context("failed to drop table")?;

        Ok(None)
    }

    fn update_schema(&self, _ctx: &MigrationContext, schema: &mut Schema) {
        schema.change_table(&self.table, |table_changes| {
            table_changes.set_removed();
        });
    }

    fn abort(&self, _ctx: &MigrationContext, _db: &mut dyn Conn) -> anyhow::Result<()> {
        Ok(())
    }
}



================================================
FILE: src/migrations/rename_table.rs
================================================
use super::{Action, MigrationContext};
use crate::{
    db::{Conn, Transaction},
    schema::Schema,
};
use anyhow::Context;
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize, Debug)]
pub struct RenameTable {
    pub table: String,
    pub new_name: String,
}

#[typetag::serde(name = "rename_table")]
impl Action for RenameTable {
    fn describe(&self) -> String {
        format!("Renaming table \"{}\" to \"{}\"", self.table, self.new_name)
    }

    fn run(
        &self,
        _ctx: &MigrationContext,
        _db: &mut dyn Conn,
        _schema: &Schema,
    ) -> anyhow::Result<()> {
        Ok(())
    }

    fn complete<'a>(
        &self,
        _ctx: &MigrationContext,
        db: &'a mut dyn Conn,
    ) -> anyhow::Result<Option<Transaction<'a>>> {
        // Rename table
        let query = format!(
            r#"
            ALTER TABLE IF EXISTS "{table}"
            RENAME TO "{new_name}"
            "#,
            table = self.table,
            new_name = self.new_name,
        );
        db.run(&query).context("failed to rename table")?;

        Ok(None)
    }

    fn update_schema(&self, _ctx: &MigrationContext, schema: &mut Schema) {
        schema.change_table(&self.table, |table_changes| {
            table_changes.set_name(&self.new_name);
        });
    }

    fn abort(&self, _ctx: &MigrationContext, _db: &mut dyn Conn) -> anyhow::Result<()> {
        Ok(())
    }
}



================================================
FILE: tests/add_column.rs
================================================
mod common;
use common::Test;

#[test]
fn add_column() {
    let mut test = Test::new("Add column");

    test.first_migration(
        r#"
        name = "create_user_table"

        [[actions]]
        type = "create_table"
        name = "users"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"

            [[actions.columns]]
            name = "name"
            type = "TEXT"
        "#,
    );

    test.second_migration(
        r#"
        name = "add_first_and_last_name_columns"

        [[actions]]
        type = "add_column"
        table = "users"

        up = "(STRING_TO_ARRAY(name, ' '))[1]"

            [actions.column]
            name = "first"
            type = "TEXT"
            nullable = false

        [[actions]]
        type = "add_column"
        table = "users"

        up = "(STRING_TO_ARRAY(name, ' '))[2]"

            [actions.column]
            name = "last"
            type = "TEXT"
            nullable = false
        "#,
    );

    test.after_first(|db| {
        // Insert some test users
        db.simple_query(
            "
            INSERT INTO users (id, name) VALUES
                (1, 'John Doe'),
                (2, 'Jane Doe');
            ",
        )
        .unwrap();
    });

    test.intermediate(|old_db, new_db| {
        // Check that the existing users have the new columns populated
        let expected = vec![("John", "Doe"), ("Jane", "Doe")];
        assert!(new_db
            .query("SELECT first, last FROM users ORDER BY id", &[],)
            .unwrap()
            .iter()
            .map(|row| (row.get("first"), row.get("last")))
            .eq(expected));

        // Insert data using old schema and make sure the new columns are populated
        old_db
            .simple_query("INSERT INTO users (id, name) VALUES (3, 'Test Testsson')")
            .unwrap();
        let (first_name, last_name): (String, String) = new_db
            .query_one("SELECT first, last from users WHERE id = 3", &[])
            .map(|row| (row.get("first"), row.get("last")))
            .unwrap();
        assert_eq!(
            ("Test", "Testsson"),
            (first_name.as_ref(), last_name.as_ref())
        );
    });

    test.after_completion(|db| {
        let expected = vec![("John", "Doe"), ("Jane", "Doe"), ("Test", "Testsson")];
        assert!(db
            .query("SELECT first, last FROM users ORDER BY id", &[],)
            .unwrap()
            .iter()
            .map(|row| (row.get("first"), row.get("last")))
            .eq(expected));
    });

    test.after_abort(|db| {
        let expected = vec![("John Doe"), ("Jane Doe"), ("Test Testsson")];
        assert!(db
            .query("SELECT name FROM users ORDER BY id", &[],)
            .unwrap()
            .iter()
            .map(|row| row.get::<'_, _, String>("name"))
            .eq(expected));
    });

    test.run()
}

#[test]
fn add_column_nullable() {
    let mut test = Test::new("Add nullable column");

    test.first_migration(
        r#"
        name = "create_users_table"

        [[actions]]
        type = "create_table"
        name = "users"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"
        "#,
    );

    test.second_migration(
        r#"
        name = "add_nullable_name_column"

        [[actions]]
        type = "add_column"
        table = "users"

            [actions.column]
            name = "name"
            type = "TEXT"
        "#,
    );

    test.after_first(|db| {
        // Insert some test values
        db.simple_query(
            "
            INSERT INTO users (id) VALUES (1), (2);
            ",
        )
        .unwrap();
    });

    test.intermediate(|old_db, new_db| {
        // Ensure existing data got updated
        let expected: Vec<Option<String>> = vec![None, None];
        assert!(new_db
            .query("SELECT name FROM users ORDER BY id", &[],)
            .unwrap()
            .iter()
            .map(|row| row.get::<_, Option<String>>("name"))
            .eq(expected));

        // Insert data using old schema and ensure new column is NULL
        old_db
            .simple_query("INSERT INTO users (id) VALUES (3)")
            .unwrap();
        let name: Option<String> = new_db
            .query_one("SELECT name from users WHERE id = 3", &[])
            .map(|row| (row.get("name")))
            .unwrap();
        assert_eq!(None, name);

        // Ensure data can be inserted against new schema
        new_db
            .simple_query("INSERT INTO users (id, name) VALUES (4, 'Test Testsson'), (5, NULL)")
            .unwrap();
    });

    test.after_completion(|db| {
        let expected: Vec<Option<String>> =
            vec![None, None, None, Some("Test Testsson".to_string()), None];
        let result: Vec<Option<String>> = db
            .query("SELECT id, name FROM users ORDER BY id", &[])
            .unwrap()
            .iter()
            .map(|row| row.get("name"))
            .collect();

        assert_eq!(result, expected);
    });

    test.run();
}

#[test]
fn add_column_with_default() {
    let mut test = Test::new("Add column with default value");

    test.first_migration(
        r#"
        name = "create_users_table"

        [[actions]]
        type = "create_table"
        name = "users"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"
        "#,
    );

    test.second_migration(
        r#"
        name = "add_name_column_with_default"

        [[actions]]
        type = "add_column"
        table = "users"

            [actions.column]
            name = "name"
            type = "TEXT"
            nullable = false
            default = "'DEFAULT'"
        "#,
    );

    test.after_first(|db| {
        // Insert some test values
        db.simple_query("INSERT INTO users (id) VALUES (1), (2)")
            .unwrap();
    });

    test.intermediate(|old_db, new_db| {
        // Ensure existing data got updated with defaults
        let expected = vec!["DEFAULT".to_string(), "DEFAULT".to_string()];
        assert!(new_db
            .query("SELECT name FROM users ORDER BY id", &[],)
            .unwrap()
            .iter()
            .map(|row| row.get::<_, String>("name"))
            .eq(expected));

        // Insert data using old schema and ensure new column gets the default value
        old_db
            .simple_query("INSERT INTO users (id) VALUES (3)")
            .unwrap();
        let name: String = new_db
            .query_one("SELECT name from users WHERE id = 3", &[])
            .map(|row| row.get("name"))
            .unwrap();
        assert_eq!("DEFAULT", name);
    });

    test.run();
}

#[test]
fn add_column_with_complex_up() {
    let mut test = Test::new("Add column complex");

    test.first_migration(
        r#"
        name = "create_tables"

        [[actions]]
        type = "create_table"
        name = "users"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"

            [[actions.columns]]
            name = "email"
            type = "TEXT"

        [[actions]]
        type = "create_table"
        name = "profiles"
        primary_key = ["user_id"]

            [[actions.columns]]
            name = "user_id"
            type = "INTEGER"
        "#,
    );

    test.second_migration(
        r#"
        name = "add_profiles_email_column"

        [[actions]]
        type = "add_column"
        table = "profiles"

            [actions.column]
            name = "email"
            type = "TEXT"
            nullable = false

            [actions.up]
            table = "users"
            value = "users.email"
            where = "profiles.user_id = users.id"
        "#,
    );

    test.after_first(|db| {
        db.simple_query("INSERT INTO users (id, email) VALUES (1, 'test@example.com')")
            .unwrap();
        db.simple_query("INSERT INTO profiles (user_id) VALUES (1)")
            .unwrap();
    });

    test.intermediate(|old_db, new_db| {
        // Ensure email was backfilled on profiles
        let email: String = new_db
            .query(
                "
                SELECT email
                FROM profiles
                WHERE user_id = 1
                ",
                &[],
            )
            .unwrap()
            .first()
            .map(|row| row.get("email"))
            .unwrap();
        assert_eq!("test@example.com", email);

        // Ensure email change in old schema is propagated to profiles table in new schema
        old_db
            .simple_query("UPDATE users SET email = 'test2@example.com' WHERE id = 1")
            .unwrap();
        let email: String = new_db
            .query(
                "
                SELECT email
                FROM profiles
                WHERE user_id = 1
                ",
                &[],
            )
            .unwrap()
            .first()
            .map(|row| row.get("email"))
            .unwrap();
        assert_eq!("test2@example.com", email);
    });

    test.run();
}



================================================
FILE: tests/add_foreign_key.rs
================================================
mod common;
use common::Test;

#[test]
fn add_foreign_key() {
    let mut test = Test::new("Add foreign key");

    test.first_migration(
        r#"
        name = "create_user_table"

        [[actions]]
        type = "create_table"
        name = "users"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"
        
        [[actions]]
        type = "create_table"
        name = "items"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"

            [[actions.columns]]
            name = "user_id"
            type = "INTEGER"
        "#,
    );

    test.second_migration(
        r#"
        name = "add_foreign_key"

        [[actions]]
        type = "add_foreign_key"
        table = "items"

            [actions.foreign_key]
            columns = ["user_id"]
            referenced_table = "users"
            referenced_columns = ["id"]
        "#,
    );

    test.after_first(|db| {
        // Insert some test users
        db.simple_query("INSERT INTO users (id) VALUES (1), (2)")
            .unwrap();
    });

    test.intermediate(|db, _| {
        // Ensure items can be inserted if they reference valid users
        db.simple_query("INSERT INTO items (id, user_id) VALUES (1, 1), (2, 2)")
            .unwrap();

        // Ensure items can't be inserted if they don't reference valid users
        let result = db.simple_query("INSERT INTO items (id, user_id) VALUES (3, 3)");
        assert!(result.is_err(), "expected insert to fail");
    });

    test.after_completion(|db| {
        // Ensure items can be inserted if they reference valid users
        db.simple_query("INSERT INTO items (id, user_id) VALUES (3, 1), (4, 2)")
            .unwrap();

        // Ensure items can't be inserted if they don't reference valid users
        let result = db.simple_query("INSERT INTO items (id, user_id) VALUES (5, 3)");
        assert!(result.is_err(), "expected insert to fail");

        // Ensure foreign key exists with the right name
        let foreign_key_name: Option<String> = db
            .query(
                "
                SELECT tc.constraint_name
                FROM information_schema.table_constraints AS tc 
                WHERE tc.constraint_type = 'FOREIGN KEY' AND tc.table_name='items';
                ",
                &[],
            )
            .unwrap()
            .first()
            .map(|row| row.get(0));
        assert_eq!(Some("items_user_id_fkey".to_string()), foreign_key_name);
    });

    test.after_abort(|db| {
        // Ensure foreign key doesn't exist
        let fk_does_not_exist = db
            .query(
                "
                SELECT tc.constraint_name
                FROM information_schema.table_constraints AS tc 
                WHERE tc.constraint_type = 'FOREIGN KEY' AND tc.table_name='items';
                ",
                &[],
            )
            .unwrap()
            .is_empty();
        assert!(fk_does_not_exist);
    });

    test.run()
}

#[test]
fn add_invalid_foreign_key() {
    let mut test = Test::new("Add invalid foreign key");

    test.first_migration(
        r#"
        name = "create_user_table"

        [[actions]]
        type = "create_table"
        name = "users"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"
        
        [[actions]]
        type = "create_table"
        name = "items"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"

            [[actions.columns]]
            name = "user_id"
            type = "INTEGER"
        "#,
    );

    test.second_migration(
        r#"
        name = "add_foreign_key"

        [[actions]]
        type = "add_foreign_key"
        table = "items"

            [actions.foreign_key]
            columns = ["user_id"]
            referenced_table = "users"
            referenced_columns = ["id"]
        "#,
    );

    test.after_first(|db| {
        // Insert some items which don't reference a valid user
        db.simple_query("INSERT INTO items (id, user_id) VALUES (1, 1), (2, 2)")
            .unwrap();
    });

    test.expect_failure();
    test.run()
}



================================================
FILE: tests/add_index.rs
================================================
mod common;
use common::Test;

#[test]
fn add_index() {
    let mut test = Test::new("Add index");

    test.first_migration(
        r#"
        name = "create_users_table"

        [[actions]]
        type = "create_table"
        name = "users"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"

            [[actions.columns]]
            name = "name"
            type = "TEXT"
        "#,
    );

    test.second_migration(
        r#"
        name = "add_users_name_index"

        [[actions]]
        type = "add_index"
        table = "users"

            [actions.index]
            name = "name_idx"
            columns = ["name"]        
        "#,
    );

    test.intermediate(|db, _| {
        // Ensure index is valid and ready
        let (is_ready, is_valid): (bool, bool) = db
            .query(
                "
                SELECT pg_index.indisready, pg_index.indisvalid
                FROM pg_catalog.pg_index
                JOIN pg_catalog.pg_class ON pg_index.indexrelid = pg_class.oid
                WHERE pg_class.relname = 'name_idx'
                ",
                &[],
            )
            .unwrap()
            .first()
            .map(|row| (row.get("indisready"), row.get("indisvalid")))
            .unwrap();

        assert!(is_ready, "expected index to be ready");
        assert!(is_valid, "expected index to be valid");
    });

    test.after_completion(|db| {
        // Ensure index is valid and ready
        let (is_ready, is_valid): (bool, bool) = db
            .query(
                "
                SELECT pg_index.indisready, pg_index.indisvalid
                FROM pg_catalog.pg_index
                JOIN pg_catalog.pg_class ON pg_index.indexrelid = pg_class.oid
                WHERE pg_class.relname = 'name_idx'
                ",
                &[],
            )
            .unwrap()
            .first()
            .map(|row| (row.get("indisready"), row.get("indisvalid")))
            .unwrap();

        assert!(is_ready, "expected index to be ready");
        assert!(is_valid, "expected index to be valid");
    });

    test.run();
}

#[test]
fn add_index_unique() {
    let mut test = Test::new("Add unique index");

    test.first_migration(
        r#"
        name = "create_users_table"

        [[actions]]
        type = "create_table"
        name = "users"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"

            [[actions.columns]]
            name = "name"
            type = "TEXT"
        "#,
    );

    test.second_migration(
        r#"
        name = "add_name_index"

        [[actions]]
        type = "add_index"
        table = "users"

            [actions.index]
            name = "name_idx"
            columns = ["name"]        
            unique = true
        "#,
    );

    test.intermediate(|db, _| {
        // Ensure index is valid, ready and unique
        let (is_ready, is_valid, is_unique): (bool, bool, bool) = db
            .query(
                "
                SELECT pg_index.indisready, pg_index.indisvalid, pg_index.indisunique
                FROM pg_catalog.pg_index
                JOIN pg_catalog.pg_class ON pg_index.indexrelid = pg_class.oid
                WHERE pg_class.relname = 'name_idx'
                ",
                &[],
            )
            .unwrap()
            .first()
            .map(|row| {
                (
                    row.get("indisready"),
                    row.get("indisvalid"),
                    row.get("indisunique"),
                )
            })
            .unwrap();

        assert!(is_ready, "expected index to be ready");
        assert!(is_valid, "expected index to be valid");
        assert!(is_unique, "expected index to be unique");
    });

    test.run();
}

#[test]
fn add_index_with_type() {
    let mut test = Test::new("Add GIN index");

    test.first_migration(
        r#"
        name = "create_users_table"

        [[actions]]
        type = "create_table"
        name = "users"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"

            [[actions.columns]]
            name = "data"
            type = "JSONB"
        "#,
    );

    test.second_migration(
        r#"
        name = "add_data_index"

        [[actions]]
        type = "add_index"
        table = "users"

            [actions.index]
            name = "data_idx"
            columns = ["data"]        
            type = "gin"
        "#,
    );

    test.intermediate(|db, _| {
        // Ensure index is valid, ready and has the right type
        let (is_ready, is_valid, index_type): (bool, bool, String) = db
            .query(
                "
                SELECT pg_index.indisready, pg_index.indisvalid, pg_am.amname
                FROM pg_catalog.pg_index
                JOIN pg_catalog.pg_class ON pg_index.indexrelid = pg_class.oid
                JOIN pg_catalog.pg_am ON pg_class.relam = pg_am.oid
                WHERE pg_class.relname = 'data_idx'
                ",
                &[],
            )
            .unwrap()
            .first()
            .map(|row| {
                (
                    row.get("indisready"),
                    row.get("indisvalid"),
                    row.get("amname"),
                )
            })
            .unwrap();

        assert!(is_ready, "expected index to be ready");
        assert!(is_valid, "expected index to be valid");
        assert_eq!("gin", index_type, "expected index type to be GIN");
    });

    test.run();
}



================================================
FILE: tests/alter_column.rs
================================================
mod common;
use common::Test;

#[test]
fn alter_column_data() {
    let mut test = Test::new("Alter column");

    test.first_migration(
        r#"
        name = "create_user_table"

        [[actions]]
        type = "create_table"
        name = "users"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"

            [[actions.columns]]
            name = "name"
            type = "TEXT"
        "#,
    );

    test.second_migration(
        r#"
        name = "uppercase_name"

        [[actions]]
        type = "alter_column"
        table = "users"
        column = "name"
        up = "UPPER(name)"
        down = "LOWER(name)"
        "#,
    );

    test.after_first(|db| {
        // Insert some test users
        db.simple_query(
            "
            INSERT INTO users (id, name) VALUES
                (1, 'john Doe'),
                (2, 'jane Doe');
            ",
        )
        .unwrap();
    });

    test.intermediate(|old_db, new_db| {
        // Check that the existing users has the altered data
        let expected = vec!["JOHN DOE", "JANE DOE"];
        assert!(new_db
            .query("SELECT name FROM users ORDER BY id", &[],)
            .unwrap()
            .iter()
            .map(|row| row.get::<_, String>("name"))
            .eq(expected));

        // Insert data using old schema and make sure the new schema gets correct values
        old_db
            .simple_query("INSERT INTO users (id, name) VALUES (3, 'test testsson')")
            .unwrap();
        let result = new_db
            .query_one("SELECT name from users WHERE id = 3", &[])
            .unwrap();
        assert_eq!("TEST TESTSSON", result.get::<_, &str>("name"));

        // Insert data using new schema and make sure the old schema gets correct values
        new_db
            .simple_query("INSERT INTO users (id, name) VALUES (4, 'TEST TESTSSON')")
            .unwrap();
        let result = old_db
            .query_one("SELECT name from users WHERE id = 4", &[])
            .unwrap();
        assert_eq!("test testsson", result.get::<_, &str>("name"));
    });

    test.run();
}

#[test]
fn alter_column_set_not_null() {
    let mut test = Test::new("Set column not null");

    test.first_migration(
        r#"
        name = "create_user_table"

        [[actions]]
        type = "create_table"
        name = "users"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"

            [[actions.columns]]
            name = "name"
            type = "TEXT"
        "#,
    );

    test.second_migration(
        r#"
        name = "set_name_not_null"

        [[actions]]
        type = "alter_column"
        table = "users"
        column = "name"
        up = "COALESCE(name, 'TEST_DEFAULT_VALUE')"

            [actions.changes]
            nullable = false
        "#,
    );

    test.after_first(|db| {
        // Insert some test users
        db.simple_query(
            "
            INSERT INTO users (id, name) VALUES
                (1, 'John Doe'),
                (2, NULL);
            ",
        )
        .unwrap();
    });

    test.intermediate(|old_db, new_db| {
        // Check that existing users got the correct values
        let expected = vec!["John Doe", "TEST_DEFAULT_VALUE"];
        assert!(new_db
            .query("SELECT name FROM users ORDER BY id", &[],)
            .unwrap()
            .iter()
            .map(|row| row.get::<_, String>("name"))
            .eq(expected));

        // Insert data using old schema and make sure the new schema gets correct values
        old_db
            .simple_query("INSERT INTO users (id, name) VALUES (3, NULL)")
            .unwrap();
        let result = new_db
            .query_one("SELECT name from users WHERE id = 3", &[])
            .unwrap();
        assert_eq!("TEST_DEFAULT_VALUE", result.get::<_, &str>("name"));

        // Insert data using new schema and make sure the old schema gets correct values
        new_db
            .simple_query("INSERT INTO users (id, name) VALUES (4, 'Jane Doe')")
            .unwrap();
        let result = old_db
            .query_one("SELECT name from users WHERE id = 4", &[])
            .unwrap();
        assert_eq!("Jane Doe", result.get::<_, &str>("name"));

        // Ensure NULL can't be inserted using the new schema
        let result = new_db.simple_query("INSERT INTO users (id, name) VALUES (5, NULL)");
        assert!(result.is_err(), "expected insert to fail");
    });

    test.after_completion(|db| {
        // Ensure NULL can't be inserted
        let result = db.simple_query("INSERT INTO users (id, name) VALUES (5, NULL)");
        assert!(result.is_err(), "expected insert to fail");
    });

    test.after_abort(|db| {
        // Ensure NULL can be inserted
        let result = db.simple_query("INSERT INTO users (id, name) VALUES (5, NULL)");
        assert!(result.is_ok(), "expected insert to succeed");
    });

    test.run();
}

#[test]
fn alter_column_set_nullable() {
    let mut test = Test::new("Set column nullable");

    test.first_migration(
        r#"
        name = "create_user_table"

        [[actions]]
        type = "create_table"
        name = "users"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"

            [[actions.columns]]
            name = "name"
            type = "TEXT"
            nullable = false
        "#,
    );

    test.second_migration(
        r#"
        name = "set_name_nullable"

        [[actions]]
        type = "alter_column"
        table = "users"
        column = "name"
        down = "COALESCE(name, 'TEST_DEFAULT_VALUE')"

            [actions.changes]
            nullable = true
        "#,
    );

    test.after_first(|db| {
        // Insert a test user
        db.simple_query(
            "
            INSERT INTO users (id, name) VALUES
                (1, 'John Doe')
            ",
        )
        .unwrap();
    });

    test.intermediate(|old_db, new_db| {
        // Insert data using new schema and make sure the old schema gets correct values
        new_db
            .simple_query("INSERT INTO users (id, name) VALUES (2, NULL)")
            .unwrap();
        let result = old_db
            .query_one("SELECT name from users WHERE id = 2", &[])
            .unwrap();
        assert_eq!("TEST_DEFAULT_VALUE", result.get::<_, &str>("name"));

        // Ensure NULL can't be inserted using the old schema
        let result = old_db.simple_query("INSERT INTO users (id, name) VALUES (3, NULL)");
        assert!(result.is_err(), "expected insert to fail");

        // Ensure NULL can be inserted using the new schema
        let result = new_db.simple_query("INSERT INTO users (id, name) VALUES (4, NULL)");
        assert!(result.is_ok(), "expected insert to succeed");
    });

    test.after_completion(|db| {
        // Ensure NULL can be inserted
        let result = db.simple_query("INSERT INTO users (id, name) VALUES (5, NULL)");
        assert!(result.is_ok(), "expected insert to succeed");
    });

    test.after_abort(|db| {
        // Ensure NULL can't be inserted
        let result = db.simple_query("INSERT INTO users (id, name) VALUES (5, NULL)");
        assert!(result.is_err(), "expected insert to fail");
    });

    test.run();
}

#[test]
fn alter_column_rename() {
    let mut test = Test::new("Rename column");

    test.first_migration(
        r#"
        name = "create_user_table"

        [[actions]]
        type = "create_table"
        name = "users"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"

            [[actions.columns]]
            name = "name"
            type = "TEXT"
        "#,
    );

    test.second_migration(
        r#"
        name = "set_name_not_null"

        [[actions]]
        type = "alter_column"
        table = "users"
        column = "name"

            [actions.changes]
            name = "full_name"
        "#,
    );

    test.after_first(|db| {
        // Insert some test data
        db.simple_query(
            "
            INSERT INTO users (id, name) VALUES
                (1, 'John Doe'),
                (2, 'Jane Doe');
            ",
        )
        .unwrap();
    });

    test.intermediate(|_, new_db| {
        // Check that existing values can be queried using new column name
        let expected = vec!["John Doe", "Jane Doe"];
        assert!(new_db
            .query("SELECT full_name FROM users ORDER BY id", &[],)
            .unwrap()
            .iter()
            .map(|row| row.get::<_, String>("full_name"))
            .eq(expected));
    });

    test.run();
}

#[test]
fn alter_column_multiple() {
    let mut test = Test::new("Alter column value multiple times");

    test.first_migration(
        r#"
        name = "create_user_table"

        [[actions]]
        type = "create_table"
        name = "users"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"

            [[actions.columns]]
            name = "counter"
            type = "INTEGER"
            nullable = false
        "#,
    );

    test.second_migration(
        r#"
        name = "increment_counter_twice"

        [[actions]]
        type = "alter_column"
        table = "users"
        column = "counter"
        up = "counter + 1"
        down = "counter - 1"

        [[actions]]
        type = "alter_column"
        table = "users"
        column = "counter"
        up = "counter + 1"
        down = "counter - 1"
        "#,
    );

    test.after_first(|db| {
        // Insert some test data
        db.simple_query(
            "
            INSERT INTO users (id, counter) VALUES
                (1, 0),
                (2, 100);
            ",
        )
        .unwrap();
    });

    test.intermediate(|old_db, new_db| {
        // Check that the existing data has been updated
        let expected = vec![2, 102];
        let results: Vec<i32> = new_db
            .query("SELECT counter FROM users ORDER BY id", &[])
            .unwrap()
            .iter()
            .map(|row| row.get::<_, i32>("counter"))
            .collect();
        assert_eq!(expected, results);

        // Update data using old schema and make sure it was updated for the new schema
        old_db
            .query("UPDATE users SET counter = 50 WHERE id = 1", &[])
            .unwrap();
        let result: i32 = new_db
            .query("SELECT counter FROM users WHERE id = 1", &[])
            .unwrap()
            .iter()
            .map(|row| row.get("counter"))
            .nth(0)
            .unwrap();
        assert_eq!(52, result);

        // Update data using new schema and make sure it was updated for the old schema
        new_db
            .query("UPDATE users SET counter = 50 WHERE id = 1", &[])
            .unwrap();
        let result: i32 = old_db
            .query("SELECT counter FROM users WHERE id = 1", &[])
            .unwrap()
            .iter()
            .map(|row| row.get("counter"))
            .nth(0)
            .unwrap();
        assert_eq!(48, result);
    });

    test.run();
}

#[test]
fn alter_column_default() {
    let mut test = Test::new("Change default value for column");

    test.first_migration(
        r#"
        name = "create_user_table"

        [[actions]]
        type = "create_table"
        name = "users"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"

            [[actions.columns]]
            name = "name"
            type = "TEXT"
            nullable = false
            default = "'DEFAULT'"
        "#,
    );

    test.second_migration(
        r#"
        name = "change_name_default"

        [[actions]]
        type = "alter_column"
        table = "users"
        column = "name"

            [actions.changes]
            default = "'NEW DEFAULT'"
        "#,
    );

    test.after_first(|db| {
        // Insert a test user
        db.simple_query(
            "
            INSERT INTO users (id) VALUES (1)
            ",
        )
        .unwrap();
    });

    test.intermediate(|old_db, new_db| {
        // Check that the existing users has the old default value
        let expected = vec!["DEFAULT"];
        assert!(new_db
            .query("SELECT name FROM users", &[],)
            .unwrap()
            .iter()
            .map(|row| row.get::<_, String>("name"))
            .eq(expected));

        // Insert data using old schema and make those get the old default value
        old_db
            .simple_query("INSERT INTO users (id) VALUES (2)")
            .unwrap();
        let result = new_db
            .query_one("SELECT name from users WHERE id = 2", &[])
            .unwrap();
        assert_eq!("DEFAULT", result.get::<_, &str>("name"));

        // Insert data using new schema and make sure it gets the new default value
        new_db
            .simple_query("INSERT INTO users (id) VALUES (3)")
            .unwrap();
        let result = old_db
            .query_one("SELECT name from users WHERE id = 3", &[])
            .unwrap();
        assert_eq!("NEW DEFAULT", result.get::<_, &str>("name"));
    });

    test.run();
}

#[test]
fn alter_column_with_index() {
    let mut test = Test::new("Alter column with index");

    test.first_migration(
        r#"
        name = "create_user_table"

        [[actions]]
        type = "create_table"
        name = "users"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"

            [[actions.columns]]
            name = "first_name"
            type = "TEXT"

            [[actions.columns]]
            name = "last_name"
            type = "TEXT"

        [[actions]]
        type = "add_index"
        table = "users"

            [actions.index]
            name = "users_name_idx"
            columns = ["first_name", "last_name"]
        "#,
    );

    test.second_migration(
        r#"
        name = "uppercase_last_name"

        [[actions]]
        type = "alter_column"
        table = "users"
        column = "last_name"
        up = "UPPER(last_name)"
        down = "LOWER(last_name)"
        "#,
    );

    test.after_completion(|db| {
        // Make sure index still exists
        let result: i64 = db
            .query(
                "
			SELECT COUNT(*)
			FROM pg_catalog.pg_index
			JOIN pg_catalog.pg_class ON pg_index.indexrelid = pg_class.oid
			WHERE pg_class.relname = 'users_name_idx'
			",
                &[],
            )
            .unwrap()
            .first()
            .map(|row| row.get(0))
            .unwrap();
        assert_eq!(1, result, "expected index to still exist");
    });

    test.run();
}

#[test]
fn alter_column_with_unique_index() {
    let mut test = Test::new("Alter column with unique index");

    test.first_migration(
        r#"
        name = "create_user_table"

        [[actions]]
        type = "create_table"
        name = "users"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"

            [[actions.columns]]
            name = "name"
            type = "TEXT"

        [[actions]]
        type = "add_index"
        table = "users"

            [actions.index]
            name = "name_idx"
            columns = ["name"]
            unique = true
        "#,
    );

    test.second_migration(
        r#"
        name = "uppercase_name"

        [[actions]]
        type = "alter_column"
        table = "users"
        column = "name"
        up = "UPPER(name)"
        down = "LOWER(name)"
        "#,
    );

    test.after_first(|db| {
        db.simple_query("INSERT INTO users (id, name) VALUES (1, 'Test')")
            .unwrap();
    });

    test.intermediate(|old_db, new_db| {
        // Try inserting a value which duplicates the uppercase value of an existing row
        let result = new_db.simple_query("INSERT INTO users (id, name) VALUES (2, 'TEST')");
        assert!(
            result.is_err(),
            "expected duplicate insert to new schema to fail"
        );

        // Try inserting a value which duplicates the lowercase value of an existing row
        new_db
            .simple_query("INSERT INTO users (id, name) VALUES (2, 'JOHN')")
            .unwrap();
        let result = old_db.simple_query("INSERT INTO users (id, name) VALUES (3, 'john')");
        assert!(
            result.is_err(),
            "expected duplicate insert to old schema to fail"
        );
    });

    test.after_completion(|db| {
        // Make sure index still exists
        let is_unique: bool = db
            .query(
                "
                SELECT pg_index.indisunique
                FROM pg_catalog.pg_index
                JOIN pg_catalog.pg_class ON pg_index.indexrelid = pg_class.oid
                WHERE pg_class.relname = 'name_idx'
                ",
                &[],
            )
            .unwrap()
            .first()
            .map(|row| row.get("indisunique"))
            .unwrap();
        assert!(is_unique, "expected index to still be unique");
    });

    test.run();
}

#[test]
fn alter_column_with_hash_index() {
    let mut test = Test::new("Alter column with custom index type");

    test.first_migration(
        r#"
        name = "create_user_table"

        [[actions]]
        type = "create_table"
        name = "users"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"

            [[actions.columns]]
            name = "name"
            type = "TEXT"

        [[actions]]
        type = "add_index"
        table = "users"

            [actions.index]
            name = "name_idx"
            columns = ["name"]
            type = "hash"
        "#,
    );

    test.second_migration(
        r#"
        name = "uppercase_name"

        [[actions]]
        type = "alter_column"
        table = "users"
        column = "name"
        up = "UPPER(name)"
        down = "LOWER(name)"
        "#,
    );

    test.after_completion(|db| {
        // Make sure index still has type GIN
        let index_type: String = db
            .query(
                "
                SELECT pg_am.amname
                FROM pg_catalog.pg_index
                JOIN pg_catalog.pg_class ON pg_index.indexrelid = pg_class.oid
                JOIN pg_catalog.pg_am ON pg_class.relam = pg_am.oid
                WHERE pg_class.relname = 'name_idx'
                ",
                &[],
            )
            .unwrap()
            .first()
            .map(|row| row.get("amname"))
            .unwrap();
        assert_eq!("hash", index_type);
    });

    test.run();
}



================================================
FILE: tests/common.rs
================================================
use colored::Colorize;
use postgres::{Client, NoTls};
use reshape::{migrations::Migration, Reshape};

pub struct Test<'a> {
    name: &'a str,
    reshape: Reshape,
    old_db: Client,
    new_db: Client,

    first_migration: Option<Migration>,
    second_migration: Option<Migration>,
    expect_failure: bool,

    clear_fn: Option<fn(&mut Client) -> ()>,
    after_first_fn: Option<fn(&mut Client) -> ()>,
    intermediate_fn: Option<fn(&mut Client, &mut Client) -> ()>,
    after_completion_fn: Option<fn(&mut Client) -> ()>,
    after_abort_fn: Option<fn(&mut Client) -> ()>,
}

impl Test<'_> {
    pub fn new<'a>(name: &'a str) -> Test<'a> {
        let connection_string = std::env::var("POSTGRES_CONNECTION_STRING")
            .unwrap_or("postgres://postgres:postgres@localhost/reshape_test".to_string());

        let old_db = Client::connect(&connection_string, NoTls).unwrap();
        let new_db = Client::connect(&connection_string, NoTls).unwrap();

        let reshape = Reshape::new(&connection_string).unwrap();

        Test {
            name,
            reshape,
            old_db,
            new_db,
            first_migration: None,
            second_migration: None,
            expect_failure: false,
            clear_fn: None,
            after_first_fn: None,
            intermediate_fn: None,
            after_completion_fn: None,
            after_abort_fn: None,
        }
    }

    pub fn first_migration(&mut self, migration: &str) -> &mut Self {
        self.first_migration = Some(Self::parse_migration(migration));
        self
    }

    #[allow(dead_code)]
    pub fn second_migration(&mut self, migration: &str) -> &mut Self {
        self.second_migration = Some(Self::parse_migration(migration));
        self
    }

    #[allow(dead_code)]
    pub fn clear(&mut self, f: fn(&mut Client) -> ()) -> &mut Self {
        self.clear_fn = Some(f);
        self
    }

    #[allow(dead_code)]
    pub fn after_first(&mut self, f: fn(&mut Client) -> ()) -> &mut Self {
        self.after_first_fn = Some(f);
        self
    }

    #[allow(dead_code)]
    pub fn intermediate(&mut self, f: fn(&mut Client, &mut Client) -> ()) -> &mut Self {
        self.intermediate_fn = Some(f);
        self
    }

    #[allow(dead_code)]
    pub fn after_completion(&mut self, f: fn(&mut Client) -> ()) -> &mut Self {
        self.after_completion_fn = Some(f);
        self
    }

    #[allow(dead_code)]
    pub fn after_abort(&mut self, f: fn(&mut Client) -> ()) -> &mut Self {
        self.after_abort_fn = Some(f);
        self
    }

    #[allow(dead_code)]
    pub fn expect_failure(&mut self) {
        self.expect_failure = true;
    }

    fn parse_migration(encoded: &str) -> Migration {
        toml::from_str(encoded).unwrap()
    }
}

enum RunType {
    Simple,
    Completion,
    Abort,
}

impl Test<'_> {
    #[allow(dead_code)]
    pub fn run(&mut self) {
        if self.second_migration.is_some() {
            // Run to completion
            print_heading(&format!("Test completion: {}", self.name));
            self.run_internal(RunType::Completion);

            // Run and abort
            print_heading(&format!("Test abort: {}", self.name));
            self.run_internal(RunType::Abort);
        } else {
            print_heading(&format!("Test: {}", self.name));
            self.run_internal(RunType::Simple);
        }
    }

    fn run_internal(&mut self, run_type: RunType) {
        print_subheading("Clearing database");
        self.reshape.remove().unwrap();

        if let Some(clear_fn) = self.clear_fn {
            clear_fn(&mut self.old_db);
        }

        // Apply first migration, will automatically complete
        print_subheading("Applying first migration");
        let first_migration = self
            .first_migration
            .as_ref()
            .expect("no starting migration set");
        self.reshape.migrate(vec![first_migration.clone()]).unwrap();

        // Update search path
        self.old_db
            .simple_query(&reshape::schema_query_for_migration(&first_migration.name))
            .unwrap();

        // Automatically complete first migration
        self.reshape.complete().unwrap();

        // Run setup function
        if let Some(after_first_fn) = self.after_first_fn {
            print_subheading("Running setup and first checks");
            after_first_fn(&mut self.old_db);
            print_success();
        }

        // Apply second migration
        if let Some(second_migration) = &self.second_migration {
            if self.expect_failure {
                print_subheading("Applying second migration (expecting failure)");
                let result = self
                    .reshape
                    .migrate(vec![first_migration.clone(), second_migration.clone()]);

                if result.is_ok() {
                    panic!("expected second migration to fail");
                }
            } else {
                print_subheading("Applying second migration");
                self.reshape
                    .migrate(vec![first_migration.clone(), second_migration.clone()])
                    .unwrap();
            }

            // Update search path
            self.new_db
                .simple_query(&reshape::schema_query_for_migration(&second_migration.name))
                .unwrap();

            if let Some(intermediate_fn) = self.intermediate_fn {
                print_subheading("Running intermediate checks");
                intermediate_fn(&mut self.old_db, &mut self.new_db);
                print_success();
            }

            match run_type {
                RunType::Completion => {
                    print_subheading("Completing");
                    self.reshape.complete().unwrap();

                    if let Some(after_completion_fn) = self.after_completion_fn {
                        print_subheading("Running post-completion checks");
                        after_completion_fn(&mut self.new_db);
                        print_success();
                    }
                }
                RunType::Abort => {
                    print_subheading("Aborting");
                    self.reshape.abort().unwrap();

                    if let Some(after_abort_fn) = self.after_abort_fn {
                        print_subheading("Running post-abort checks");
                        after_abort_fn(&mut self.old_db);
                        print_success();
                    }
                }
                _ => {}
            }
        }

        print_subheading("Checking cleanup");
        assert_cleaned_up(&mut self.new_db);
        print_success();
    }
}

fn print_heading(text: &str) {
    let delimiter = std::iter::repeat("=").take(80).collect::<String>();

    println!();
    println!();
    println!("{}", delimiter.blue().bold());
    println!("{}", add_spacer(text, "=").blue().bold());
    println!("{}", delimiter.blue().bold());
}

fn print_subheading(text: &str) {
    println!();
    println!("{}", add_spacer(text, "=").blue());
}

fn print_success() {
    println!("{}", add_spacer("Success", "=").green());
}

fn add_spacer(text: &str, char: &str) -> String {
    const TARGET_WIDTH: usize = 80;
    let num_of_chars = (TARGET_WIDTH - text.len() - 2) / 2;
    let spacer = std::iter::repeat(char)
        .take(num_of_chars)
        .collect::<String>();

    let extra = if text.len() % 2 == 0 { "" } else { char };

    format!("{spacer} {text} {spacer}{extra}", spacer = spacer)
}

pub fn assert_cleaned_up(db: &mut Client) {
    // Make sure no temporary columns remain
    let temp_columns: Vec<String> = db
        .query(
            "
            SELECT column_name
            FROM information_schema.columns
            WHERE table_schema = 'public'
            AND column_name LIKE '__reshape%'
            ",
            &[],
        )
        .unwrap()
        .iter()
        .map(|row| row.get(0))
        .collect();

    assert!(
        temp_columns.is_empty(),
        "expected no temporary columns to exist, found: {}",
        temp_columns.join(", ")
    );

    // Make sure no triggers remain
    let triggers: Vec<String> = db
        .query(
            "
            SELECT trigger_name
            FROM information_schema.triggers
            WHERE trigger_schema = 'public'
            AND trigger_name LIKE '__reshape%'
            ",
            &[],
        )
        .unwrap()
        .iter()
        .map(|row| row.get(0))
        .collect();

    assert!(
        triggers.is_empty(),
        "expected no triggers to exist, found: {}",
        triggers.join(", ")
    );

    // Make sure no functions remain
    let functions: Vec<String> = db
        .query(
            "
            SELECT routine_name
            FROM information_schema.routines
            WHERE routine_schema = 'public'
            AND routine_name LIKE '__reshape%'
            ",
            &[],
        )
        .unwrap()
        .iter()
        .map(|row| row.get(0))
        .collect();

    assert!(
        functions.is_empty(),
        "expected no functions to exist, found: {}",
        functions.join(", ")
    );
}



================================================
FILE: tests/complex.rs
================================================
mod common;
use common::Test;

#[test]
fn move_column_between_tables() {
    let mut test = Test::new("Move column between tables");

    test.first_migration(
        r#"
        name = "create_tables"

        [[actions]]
        type = "create_table"
        name = "users"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"

            [[actions.columns]]
            name = "email"
            type = "TEXT"

        [[actions]]
        type = "create_table"
        name = "profiles"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"

            [[actions.columns]]
            name = "user_id"
            type = "INTEGER"
            nullable = false
        "#,
    );

    test.second_migration(
        r#"
        name = "move_email_column"

        [[actions]]
        type = "add_column"
        table = "profiles"

            [actions.column]
            name = "email"
            type = "TEXT"
            nullable = false

            # When `users` is updated in the old schema, we write the email value to `profiles`
            # When `profiles` is updated in the old schema, the equivalent `users.email` will also be updated
            [actions.up]
            table = "users"
            value = "users.email"
            where = "profiles.user_id = users.id"

        [[actions]]
        type = "remove_column"
        table = "users"
        column = "email"

            # When `profiles` is changed in the new schema, we write the email address back to the removed column
            [actions.down]
            table = "profiles"
            value = "profiles.email"
            where = "users.id = profiles.user_id"
        "#,
    );

    test.after_first(|db| {
        db.simple_query("INSERT INTO users (id, email) VALUES (1, 'test1@test.com')")
            .unwrap();
        db.simple_query("INSERT INTO users (id, email) VALUES (2, 'test2@test.com')")
            .unwrap();

        db.simple_query("INSERT INTO profiles (id, user_id) VALUES (1, 1)")
            .unwrap();
        db.simple_query("INSERT INTO profiles (id, user_id) VALUES (2, 2)")
            .unwrap();
    });

    test.intermediate(|old_db, new_db| {
        // Ensure emails were backfilled into profiles
        let profiles_emails: Vec<String> = new_db
            .query(
                r#"
                SELECT email
                FROM profiles
                ORDER BY id
                "#,
                &[],
            )
            .unwrap()
            .iter()
            .map(|row| row.get("email"))
            .collect();
        assert_eq!(vec!("test1@test.com", "test2@test.com"), profiles_emails);

        // Ensure insert in old schema updates new
        old_db
            .simple_query("INSERT INTO users (id, email) VALUES (3, 'test3@test.com')")
            .unwrap();
        old_db
            .simple_query("INSERT INTO profiles (id, user_id) VALUES (3, 3)")
            .unwrap();
        let new_email: String = new_db
            .query("SELECT email FROM profiles WHERE id = 3", &[])
            .unwrap()
            .first()
            .map(|row| row.get("email"))
            .unwrap();
        assert_eq!("test3@test.com", new_email);

        // Ensure updates in old schema updates new
        old_db
            .simple_query("UPDATE users SET email = 'test3+updated@test.com' WHERE id = 3")
            .unwrap();
        let new_email: String = new_db
            .query("SELECT email FROM profiles WHERE id = 3", &[])
            .unwrap()
            .first()
            .map(|row| row.get("email"))
            .unwrap();
        assert_eq!("test3+updated@test.com", new_email);

        // Ensure insert in new schema updates old
        new_db
            .simple_query(
                "INSERT INTO profiles (id, user_id, email) VALUES (4, 4, 'test4@test.com')",
            )
            .unwrap();
        new_db
            .simple_query("INSERT INTO users (id) VALUES (4)")
            .unwrap();
        let old_email: String = old_db
            .query("SELECT email FROM users WHERE id = 4", &[])
            .unwrap()
            .first()
            .map(|row| row.get("email"))
            .unwrap();
        assert_eq!("test4@test.com", old_email);

        // Ensure update in new schema updates old
        new_db
            .simple_query("UPDATE profiles SET email = 'test4+updated@test.com' WHERE id = 4")
            .unwrap();
        let old_email: String = old_db
            .query("SELECT email FROM users WHERE id = 4", &[])
            .unwrap()
            .first()
            .map(|row| row.get("email"))
            .unwrap();
        assert_eq!("test4+updated@test.com", old_email);
    });

    test.run();
}

#[test]
fn extract_relation_into_new_table() {
    let mut test = Test::new("Extract relation into new table");

    test.first_migration(
        r#"
        name = "create_tables"

        [[actions]]
        type = "create_table"
        name = "accounts"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"

        [[actions]]
        type = "create_table"
        name = "users"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"

            [[actions.columns]]
            name = "account_id"
            type = "INTEGER"
            nullable = false

            [[actions.columns]]
            name = "account_role"
            type = "TEXT"
            nullable = false
        "#,
    );

    test.second_migration(
        r#"
        name = "add_account_user_connection"

        [[actions]]
        type = "create_table"
        name = "user_account_connections"
        primary_key = ["account_id", "user_id"]

            [[actions.columns]]
            name = "account_id"
            type = "INTEGER"

            [[actions.columns]]
            name = "user_id"
            type = "INTEGER"

            [[actions.columns]]
            name = "role"
            type = "TEXT"
            nullable = false

            [actions.up]
            table = "users"
            values = { user_id = "id", account_id = "account_id", role = "UPPER(account_role)" }
            where = "user_account_connections.user_id = users.id"

        [[actions]]
        type = "remove_column"
        table = "users"
        column = "account_id"

            [actions.down]
            table = "user_account_connections"
            value = "user_account_connections.account_id"
            where = "users.id = user_account_connections.user_id"

        [[actions]]
        type = "remove_column"
        table = "users"
        column = "account_role"

            [actions.down]
            table = "user_account_connections"
            value = "LOWER(user_account_connections.role)"
            where = "users.id = user_account_connections.user_id"
        "#,
    );

    test.after_first(|db| {
        db.simple_query("INSERT INTO accounts (id) VALUES (1)")
            .unwrap();
        db.simple_query("INSERT INTO users (id, account_id, account_role) VALUES (1, 1, 'admin')")
            .unwrap();
    });

    test.intermediate(|old_db, new_db| {
        // Ensure connections was backfilled
        let rows: Vec<(i32, i32, String)> = new_db
            .query(
                "
                SELECT account_id, user_id, role
                FROM user_account_connections
                ",
                &[],
            )
            .unwrap()
            .iter()
            .map(|row| (row.get("account_id"), row.get("user_id"), row.get("role")))
            .collect();
        assert_eq!(1, rows.len());

        let row = rows.first().unwrap();
        assert_eq!(1, row.0);
        assert_eq!(1, row.1);
        assert_eq!("ADMIN", row.2);

        // Ensure inserted user in old schema creates a new connection
        old_db
            .simple_query(
                "INSERT INTO users (id, account_id, account_role) VALUES (2, 1, 'developer')",
            )
            .unwrap();
        assert!(
            new_db
                .query(
                    "
                    SELECT account_id, user_id, role
                    FROM user_account_connections
                    WHERE account_id = 1 AND user_id = 2 AND role = 'DEVELOPER'
                    ",
                    &[],
                )
                .unwrap()
                .len()
                == 1
        );

        // Ensure NOT NULL constraint still applies to old schema
        let result = old_db
            .simple_query(
                "INSERT INTO users (id, account_id, account_role) VALUES (2, NULL, 'developer')",
            );
            assert!(result.is_err());

        // Ensure updated user role in old schema updates connection in new schema
        old_db
            .simple_query("UPDATE users SET account_role = 'admin' WHERE id = 2")
            .unwrap();
        assert!(
            new_db
                .query(
                    "
                    SELECT account_id, user_id, role
                    FROM user_account_connections
                    WHERE account_id = 1 AND user_id = 2 AND role = 'ADMIN'
                    ",
                    &[],
                )
                .unwrap()
                .len()
                == 1
        );

        // Ensure updated connection in new schema updates old schema user
        new_db
            .simple_query(
                "UPDATE user_account_connections SET role = 'DEVELOPER' WHERE account_id = 1 AND user_id = 2",
            )
            .unwrap();
        assert!(
            old_db
                .query(
                    "
                    SELECT id
                    FROM users
                    WHERE id = 2 AND account_id = 1 AND account_role = 'developer'
                    ",
                    &[],
                )
                .unwrap()
                .len()
                == 1
        );

        // Ensure insert of user with connection through new schema updates user in old schema
        new_db
            .simple_query(
                r#"
                BEGIN;
                INSERT INTO users (id) VALUES (3);
                INSERT INTO user_account_connections (user_id, account_id, role) VALUES (3, 1, 'DEVELOPER');
                COMMIT;
                "#,
            )
            .unwrap();
        new_db
            .simple_query(
                "",
            )
            .unwrap();
        assert!(
            old_db
                .query(
                    "
                    SELECT id
                    FROM users
                    WHERE id = 3 AND account_id = 1 AND account_role = 'developer'
                    ",
                    &[],
                )
                .unwrap()
                .len()
                == 1
        );
    });

    test.run();
}



================================================
FILE: tests/create_enum.rs
================================================
mod common;
use common::Test;

#[test]
fn create_enum() {
    let mut test = Test::new("Create enum");

    test.first_migration(
        r#"
		name = "create_enum_and_table"

		[[actions]]
		type = "create_enum"
		name = "mood"
		values = ["happy", "ok", "sad"]

		[[actions]]
		type = "create_table"
		name = "updates"
		primary_key = ["id"]

			[[actions.columns]]
			name = "id"
			type = "INTEGER"

			[[actions.columns]]
			name = "status"
			type = "mood"
		"#,
    );

    test.after_first(|db| {
        // Valid enum values should succeed
        db.simple_query(
            "INSERT INTO updates (id, status) VALUES (1, 'happy'), (2, 'ok'), (3, 'sad')",
        )
        .unwrap();
    });

    test.run();
}



================================================
FILE: tests/create_table.rs
================================================
mod common;
use common::Test;

#[test]
fn create_table() {
    let mut test = Test::new("Create table");

    test.first_migration(
        r#"
        name = "create_users_table"
        
        [[actions]]
        type = "create_table"
        name = "users"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"
            generated = "ALWAYS AS IDENTITY"

            [[actions.columns]]
            name = "name"
            type = "TEXT"

            [[actions.columns]]
            name = "created_at"
            type = "TIMESTAMP"
            nullable = false
            default = "NOW()"
        "#,
    );

    test.after_first(|db| {
        // Ensure table was created
        let result = db
            .query_opt(
                "
                SELECT table_name
                FROM information_schema.tables
                WHERE table_name = 'users' AND table_schema = 'public'",
                &[],
            )
            .unwrap();
        assert!(result.is_some());

        // Ensure table has the right columns
        let result = db
            .query(
                "
                SELECT column_name, column_default, is_nullable, data_type
                FROM information_schema.columns
                WHERE table_name = 'users' AND table_schema = 'public'
                ORDER BY ordinal_position",
                &[],
            )
            .unwrap();

        // id column
        let id_row = &result[0];
        assert_eq!("id", id_row.get::<_, String>("column_name"));
        assert!(id_row.get::<_, Option<String>>("column_default").is_none());
        assert_eq!("NO", id_row.get::<_, String>("is_nullable"));
        assert_eq!("integer", id_row.get::<_, String>("data_type"));

        // name column
        let name_row = &result[1];
        assert_eq!("name", name_row.get::<_, String>("column_name"));
        assert!(name_row
            .get::<_, Option<String>>("column_default")
            .is_none());
        assert_eq!("YES", name_row.get::<_, String>("is_nullable"));
        assert_eq!("text", name_row.get::<_, String>("data_type"));

        // created_at column
        let created_at_column = &result[2];
        assert_eq!(
            "created_at",
            created_at_column.get::<_, String>("column_name")
        );
        assert!(created_at_column
            .get::<_, Option<String>>("column_default")
            .is_some());
        assert_eq!("NO", created_at_column.get::<_, String>("is_nullable"));
        assert_eq!(
            "timestamp without time zone",
            created_at_column.get::<_, String>("data_type")
        );

        // Ensure the primary key has the right columns
        let primary_key_columns: Vec<String> = db
            .query(
                "
                SELECT a.attname AS column
                FROM pg_index i
                JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey)
                JOIN pg_class t ON t.oid = i.indrelid
                WHERE t.relname = 'users' AND i.indisprimary
                ",
                &[],
            )
            .unwrap()
            .iter()
            .map(|row| row.get("column"))
            .collect();

        assert_eq!(vec!["id"], primary_key_columns);
    });

    test.run();
}

#[test]
fn create_table_with_foreign_keys() {
    let mut test = Test::new("Create table");

    test.first_migration(
        r#"
        name = "create_users_table"
        
        [[actions]]
        type = "create_table"
        name = "users"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"
            generated = "ALWAYS AS IDENTITY"

            [[actions.columns]]
            name = "name"
            type = "TEXT"

            [[actions.columns]]
            name = "created_at"
            type = "TIMESTAMP"
            nullable = false
            default = "NOW()"

        [[actions]]
        type = "create_table"
        name = "items"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"

            [[actions.columns]]
            name = "user_id"
            type = "INTEGER"
            nullable = false

            [[actions.foreign_keys]]
            columns = ["user_id"]
            referenced_table = "users"
            referenced_columns = ["id"]
        "#,
    );

    test.after_first(|db| {
        let foreign_key_columns: Vec<(String, String, String)> = db
            .query(
                "
                SELECT
                    kcu.column_name, 
                    ccu.table_name AS foreign_table_name,
                    ccu.column_name AS foreign_column_name 
                FROM 
                    information_schema.table_constraints AS tc 
                    JOIN information_schema.key_column_usage AS kcu
                    ON tc.constraint_name = kcu.constraint_name
                    AND tc.table_schema = kcu.table_schema
                    JOIN information_schema.constraint_column_usage AS ccu
                    ON ccu.constraint_name = tc.constraint_name
                    AND ccu.table_schema = tc.table_schema
                WHERE tc.constraint_type = 'FOREIGN KEY' AND tc.table_name='items';
                ",
                &[],
            )
            .unwrap()
            .iter()
            .map(|row| {
                (
                    row.get("column_name"),
                    row.get("foreign_table_name"),
                    row.get("foreign_column_name"),
                )
            })
            .collect();

        assert_eq!(
            vec![("user_id".to_string(), "users".to_string(), "id".to_string())],
            foreign_key_columns
        );
    });

    test.run();
}



================================================
FILE: tests/custom.rs
================================================
mod common;
use common::Test;

#[test]
fn custom_enable_extension() {
    let mut test = Test::new("Custom migration");

    test.clear(|db| {
        db.simple_query(
            "
			DROP EXTENSION IF EXISTS bloom;
			DROP EXTENSION IF EXISTS btree_gin;
			DROP EXTENSION IF EXISTS btree_gist;
            ",
        )
        .unwrap();
    });

    test.first_migration(
        r#"
		name = "empty_migration"

		[[actions]]
		type = "custom"
		"#,
    );

    test.second_migration(
        r#"
		name = "enable_extensions"

		[[actions]]
		type = "custom"

		start = """
			CREATE EXTENSION IF NOT EXISTS bloom;
			CREATE EXTENSION IF NOT EXISTS btree_gin;
		"""

        complete = "CREATE EXTENSION IF NOT EXISTS btree_gist"

		abort = """
			DROP EXTENSION IF EXISTS bloom;
			DROP EXTENSION IF EXISTS btree_gin;
		"""
		"#,
    );

    test.intermediate(|db, _| {
        let bloom_activated = !db
            .query("SELECT * FROM pg_extension WHERE extname = 'bloom'", &[])
            .unwrap()
            .is_empty();
        assert!(bloom_activated);

        let btree_gin_activated = !db
            .query(
                "SELECT * FROM pg_extension WHERE extname = 'btree_gin'",
                &[],
            )
            .unwrap()
            .is_empty();
        assert!(btree_gin_activated);

        let btree_gist_activated = !db
            .query(
                "SELECT * FROM pg_extension WHERE extname = 'btree_gist'",
                &[],
            )
            .unwrap()
            .is_empty();
        assert!(!btree_gist_activated);
    });

    test.after_completion(|db| {
        let bloom_activated = !db
            .query("SELECT * FROM pg_extension WHERE extname = 'bloom'", &[])
            .unwrap()
            .is_empty();
        assert!(bloom_activated);

        let btree_gin_activated = !db
            .query(
                "SELECT * FROM pg_extension WHERE extname = 'btree_gin'",
                &[],
            )
            .unwrap()
            .is_empty();
        assert!(btree_gin_activated);

        let btree_gist_activated = !db
            .query(
                "SELECT * FROM pg_extension WHERE extname = 'btree_gist'",
                &[],
            )
            .unwrap()
            .is_empty();
        assert!(btree_gist_activated);
    });

    test.after_abort(|db| {
        let bloom_activated = !db
            .query("SELECT * FROM pg_extension WHERE extname = 'bloom'", &[])
            .unwrap()
            .is_empty();
        assert!(!bloom_activated);

        let btree_gin_activated = !db
            .query(
                "SELECT * FROM pg_extension WHERE extname = 'btree_gin'",
                &[],
            )
            .unwrap()
            .is_empty();
        assert!(!btree_gin_activated);

        let btree_gist_activated = !db
            .query(
                "SELECT * FROM pg_extension WHERE extname = 'btree_gist'",
                &[],
            )
            .unwrap()
            .is_empty();
        assert!(!btree_gist_activated);
    });

    test.run();
}



================================================
FILE: tests/failure.rs
================================================
mod common;
use common::Test;

#[test]
fn invalid_migration() {
    let mut test = Test::new("Invalid migration");

    test.first_migration(
        r#"
        name = "invalid_migration"

        [[actions]]
        type = "create_table"
        name = "users"
        primary_key = ["id"]

            [[actions.columns]]    
            name = "id"
            type = "INTEGER"
        "#,
    );

    test.second_migration(
        r#"
        name = "add_invalid_column"

        [[actions]]
        type = "add_column"
        table = "users"

        up = "INVALID SQL"

            [actions.column]
            name = "first"
            type = "TEXT"
        "#,
    );

    // Insert a test user
    test.after_first(|db| {
        db.simple_query(
            "
            INSERT INTO users (id) VALUES (1)
            ",
        )
        .unwrap();
    });

    test.expect_failure();
    test.run();
}



================================================
FILE: tests/remove_column.rs
================================================
mod common;
use common::Test;

#[test]
fn remove_column() {
    let mut test = Test::new("Remove column");

    test.first_migration(
        r#"
        name = "create_user_table"

        [[actions]]
        type = "create_table"
        name = "users"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"

            [[actions.columns]]
            name = "name"
            type = "TEXT"
        "#,
    );

    test.second_migration(
        r#"
        name = "remove_name_column"

        [[actions]]
        type = "remove_column"
        table = "users"
        column = "name"
        down = "'TEST_DOWN_VALUE'"
        "#,
    );

    test.intermediate(|old_db, new_db| {
        // Insert using old schema and ensure it can be retrieved through new schema
        old_db
            .simple_query("INSERT INTO users(id, name) VALUES (1, 'John Doe')")
            .unwrap();
        let results = new_db
            .query("SELECT id FROM users WHERE id = 1", &[])
            .unwrap();
        assert_eq!(1, results.len());
        assert_eq!(1, results[0].get::<_, i32>("id"));

        // Ensure the name column is not accesible through the new schema
        assert!(new_db.query("SELECT id, name FROM users", &[]).is_err());

        // Insert using new schema and ensure the down function is correctly applied
        new_db
            .simple_query("INSERT INTO users(id) VALUES (2)")
            .unwrap();
        let result = old_db
            .query_opt("SELECT name FROM users WHERE id = 2", &[])
            .unwrap();
        assert_eq!(
            Some("TEST_DOWN_VALUE"),
            result.as_ref().map(|row| row.get("name"))
        );
    });

    test.run();
}

#[test]
fn remove_column_with_index() {
    let mut test = Test::new("Remove column");

    test.first_migration(
        r#"
        name = "create_user_table"

        [[actions]]
        type = "create_table"
        name = "users"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"

            [[actions.columns]]
            name = "name"
            type = "TEXT"
        
        [[actions]]
        type = "add_index"
        table = "users"

            [actions.index]
            name = "name_idx"
            columns = ["name"]
        "#,
    );

    test.second_migration(
        r#"
        name = "remove_name_column"

        [[actions]]
        type = "remove_column"
        table = "users"
        column = "name"
        down = "'TEST_DOWN_VALUE'"
        "#,
    );

    test.after_completion(|db| {
        // Ensure index has been removed after the migration is complete
        let count: i64 = db
            .query(
                "
                SELECT COUNT(*)
                FROM pg_catalog.pg_index
                JOIN pg_catalog.pg_class ON pg_index.indexrelid = pg_class.oid
                WHERE pg_class.relname = 'name_idx'
                ",
                &[],
            )
            .unwrap()
            .first()
            .map(|row| row.get(0))
            .unwrap();

        assert_eq!(0, count, "expected index to not exist");
    });

    test.run();
}

#[test]
fn remove_column_with_complex_down() {
    let mut test = Test::new("Remove column complex");

    test.first_migration(
        r#"
        name = "create_tables"

        [[actions]]
        type = "create_table"
        name = "users"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"

            [[actions.columns]]
            name = "email"
            type = "TEXT"

        [[actions]]
        type = "create_table"
        name = "profiles"
        primary_key = ["user_id"]

            [[actions.columns]]
            name = "user_id"
            type = "INTEGER"

            [[actions.columns]]
            name = "email"
            type = "TEXT"
        "#,
    );

    test.second_migration(
        r#"
        name = "remove_users_email_column"

        [[actions]]
        type = "remove_column"
        table = "users"
        column = "email"

            [actions.down]
            table = "profiles"
            value = "profiles.email"
            where = "users.id = profiles.user_id"
        "#,
    );

    test.after_first(|db| {
        db.simple_query("INSERT INTO users (id, email) VALUES (1, 'test@example.com')")
            .unwrap();
        db.simple_query("INSERT INTO profiles (user_id, email) VALUES (1, 'test@example.com')")
            .unwrap();
    });

    test.intermediate(|old_db, new_db| {
        new_db
            .simple_query("UPDATE profiles SET email = 'test2@example.com' WHERE user_id = 1")
            .unwrap();

        // Ensure new email was propagated to users table in old schema
        let email: String = old_db
            .query(
                "
                SELECT email
                FROM users
                WHERE id = 1
                ",
                &[],
            )
            .unwrap()
            .first()
            .map(|row| row.get("email"))
            .unwrap();
        assert_eq!("test2@example.com", email);
    });

    test.run();
}



================================================
FILE: tests/remove_enum.rs
================================================
mod common;
use common::Test;

#[test]
fn remove_enum() {
    let mut test = Test::new("Remove enum");

    test.first_migration(
        r#"
		name = "create_enum"

		[[actions]]
		type = "create_enum"
		name = "mood"
		values = ["happy", "ok", "sad"]
		"#,
    );

    test.second_migration(
        r#"
		name = "remove_enum"

		[[actions]]
		type = "remove_enum"
		enum = "mood"
		"#,
    );

    test.after_first(|db| {
        // Ensure enum was created
        let enum_exists = !db
            .query(
                "
				SELECT typname
				FROM pg_catalog.pg_type
				WHERE typcategory = 'E'
				AND typname = 'mood'
				",
                &[],
            )
            .unwrap()
            .is_empty();

        assert!(enum_exists, "expected mood enum to have been created");
    });

    test.after_completion(|db| {
        // Ensure enum was removed after completion
        let enum_does_not_exist = db
            .query(
                "
				SELECT typname
				FROM pg_catalog.pg_type
				WHERE typcategory = 'E'
				AND typname = 'mood'
				",
                &[],
            )
            .unwrap()
            .is_empty();

        assert!(
            enum_does_not_exist,
            "expected mood enum to have been removed"
        );
    });

    test.run();
}



================================================
FILE: tests/remove_foreign_key.rs
================================================
mod common;
use common::Test;

#[test]
fn remove_foreign_key() {
    let mut test = Test::new("Remove foreign key");

    test.first_migration(
        r#"
        name = "create_tables"

        [[actions]]
        type = "create_table"
        name = "users"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"
        
        [[actions]]
        type = "create_table"
        name = "items"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"

            [[actions.columns]]
            name = "user_id"
            type = "INTEGER"

            [[actions.foreign_keys]]
            columns = ["user_id"]
            referenced_table = "users"
            referenced_columns = ["id"]
        "#,
    );

    test.second_migration(
        r#"
        name = "remove_foreign_key"

        [[actions]]
        type = "remove_foreign_key"
        table = "items"
        foreign_key = "items_user_id_fkey"
        "#,
    );

    test.after_first(|db| {
        // Insert some test users
        db.simple_query("INSERT INTO users (id) VALUES (1), (2)")
            .unwrap();
    });

    test.intermediate(|old_db, new_db| {
        // Ensure items can't be inserted if they don't reference valid users
        // The foreign key is only removed when the migration is completed so
        // it should still be enforced for the new and old schema.
        let result = old_db.simple_query("INSERT INTO items (id, user_id) VALUES (3, 3)");
        assert!(
            result.is_err(),
            "expected insert against old schema to fail"
        );

        let result = new_db.simple_query("INSERT INTO items (id, user_id) VALUES (3, 3)");
        assert!(
            result.is_err(),
            "expected insert against new schema to fail"
        );
    });

    test.after_completion(|db| {
        // Ensure items can be inserted even if they don't reference valid users
        let result = db
            .simple_query("INSERT INTO items (id, user_id) VALUES (5, 3)")
            .map(|_| ());
        assert!(
            result.is_ok(),
            "expected insert to not fail, got {:?}",
            result
        );

        // Ensure foreign key doesn't exist
        let foreign_keys = db
            .query(
                "
                SELECT tc.constraint_name
                FROM information_schema.table_constraints AS tc 
                WHERE tc.constraint_type = 'FOREIGN KEY' AND tc.table_name='items';
                ",
                &[],
            )
            .unwrap();
        assert!(
            foreign_keys.is_empty(),
            "expected no foreign keys to exist on items table"
        );
    });

    test.after_abort(|db| {
        // Ensure items can't be inserted if they don't reference valid users
        let result = db.simple_query("INSERT INTO items (id, user_id) VALUES (3, 3)");
        assert!(result.is_err(), "expected insert to fail");

        // Ensure foreign key still exists
        let fk_exists = !db
            .query(
                "
                SELECT tc.constraint_name
                FROM information_schema.table_constraints AS tc 
                WHERE tc.constraint_type = 'FOREIGN KEY' AND tc.table_name='items';
                ",
                &[],
            )
            .unwrap()
            .is_empty();
        assert!(fk_exists);
    });

    test.run()
}



================================================
FILE: tests/remove_index.rs
================================================
mod common;
use common::Test;

#[test]
fn remove_index() {
    let mut test = Test::new("Remove index");

    test.first_migration(
        r#"
        name = "create_users_table"

        [[actions]]
        type = "create_table"
        name = "users"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"

            [[actions.columns]]
            name = "name"
            type = "TEXT"

        [[actions]]
        type = "add_index"
        table = "users"

            [actions.index]
            name = "name_idx"
            columns = ["name"]        
        "#,
    );

    test.second_migration(
        r#"
        name = "remove_name_index"

        [[actions]]
        type = "remove_index"
        index = "name_idx"
        "#,
    );

    test.intermediate(|db, _| {
        // Ensure index is still valid and ready during the migration
        let result: Vec<(bool, bool)> = db
            .query(
                "
                SELECT pg_index.indisready, pg_index.indisvalid
                FROM pg_catalog.pg_index
                JOIN pg_catalog.pg_class ON pg_index.indexrelid = pg_class.oid
                WHERE pg_class.relname = 'name_idx'
                ",
                &[],
            )
            .unwrap()
            .iter()
            .map(|row| (row.get("indisready"), row.get("indisvalid")))
            .collect();

        assert_eq!(vec![(true, true)], result);
    });

    test.after_completion(|db| {
        // Ensure index has been removed after the migration is complete
        let count: i64 = db
            .query(
                "
                SELECT COUNT(*)
                FROM pg_catalog.pg_index
                JOIN pg_catalog.pg_class ON pg_index.indexrelid = pg_class.oid
                WHERE pg_class.relname = 'name_idx'
                ",
                &[],
            )
            .unwrap()
            .first()
            .map(|row| row.get(0))
            .unwrap();

        assert_eq!(0, count, "expected index to not exist");
    });

    test.run();
}



================================================
FILE: tests/remove_table.rs
================================================
mod common;
use common::Test;

#[test]
fn remove_table() {
    let mut test = Test::new("Remove table");

    test.first_migration(
        r#"
        name = "create_users_table"
        
        [[actions]]
        type = "create_table"
        name = "users"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"
        "#,
    );

    test.second_migration(
        r#"
        name = "remove_users_table"

        [[actions]]
        type = "remove_table"
        table = "users"
        "#,
    );

    test.intermediate(|old_db, new_db| {
        // Make sure inserts work against the old schema
        old_db
            .simple_query("INSERT INTO users(id) VALUES (1)")
            .unwrap();

        // Ensure the table is not accessible through the new schema
        assert!(new_db.query("SELECT id FROM users", &[]).is_err());
    });

    test.run();
}



================================================
FILE: tests/rename_table.rs
================================================
mod common;
use common::Test;

#[test]
fn rename_table() {
    let mut test = Test::new("Rename table");

    test.first_migration(
        r#"
        name = "create_users_table"
        
        [[actions]]
        type = "create_table"
        name = "users"
        primary_key = ["id"]

            [[actions.columns]]
            name = "id"
            type = "INTEGER"
        "#,
    );

    test.second_migration(
        r#"
        name = "rename_users_table_to_customers"

        [[actions]]
        type = "rename_table"
        table = "users"
        new_name = "customers"
        "#,
    );

    test.intermediate(|old_db, new_db| {
        // Make sure inserts work using both the old and new name
        old_db
            .simple_query("INSERT INTO users(id) VALUES (1)")
            .unwrap();
        new_db
            .simple_query("INSERT INTO customers(id) VALUES (2)")
            .unwrap();

        // Ensure the table can be queried using both the old and new name
        let expected: Vec<i32> = vec![1, 2];
        assert_eq!(
            expected,
            old_db
                .query("SELECT id FROM users ORDER BY id", &[])
                .unwrap()
                .iter()
                .map(|row| row.get::<_, i32>("id"))
                .collect::<Vec<i32>>()
        );
        assert_eq!(
            expected,
            new_db
                .query("SELECT id FROM customers ORDER BY id", &[])
                .unwrap()
                .iter()
                .map(|row| row.get::<_, i32>("id"))
                .collect::<Vec<i32>>()
        );

        // Ensure the table can't be queried using the wrong name for the schema
        assert!(old_db.simple_query("SELECT id FROM customers").is_err());
        assert!(new_db.simple_query("SELECT id FROM users").is_err());
    });

    test.run();
}



================================================
FILE: .github/workflows/lint.yaml
================================================
name: Lint

on:
  push:
    branches: [ main ]
  pull_request:
    branches:
    - main

env:
  CARGO_TERM_COLOR: always

jobs:
  lint:
    runs-on: ubuntu-latest
    steps:
    - name: Checkout
      uses: actions/checkout@v4
    - name: Select Rust toolchain with Clippy
      uses: actions-rs/toolchain@v1
      with:
          toolchain: stable
          components: clippy
          override: true
    - name: Use cache for Rust dependencies
      uses: Swatinem/rust-cache@v2
    - name: Lint using Clippy
      run: cargo clippy



================================================
FILE: .github/workflows/publish-crate.yaml
================================================
name: Publish crate

on: workflow_dispatch

jobs:
  publish-crate:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout
        uses: actions/checkout@v2
      - name: Select Rust toolchain
        uses: actions-rs/toolchain@v1
        with:
          toolchain: stable
      - name: Publish to crates.io
        uses: actions-rs/cargo@v1
        with:
          command: publish
        env:
          CARGO_REGISTRY_TOKEN: ${{ secrets.CRATES_IO_TOKEN }}



================================================
FILE: .github/workflows/publish-docker.yaml
================================================
name: Publish Docker image

on:
  workflow_dispatch:
    inputs:
      version:
        description: "Version (without 'v' prefix)"
        required: true
        type: string

jobs:
  publish-docker:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout
        uses: actions/checkout@v2
      - name: Log in to Docker
        env:
          DOCKER_PASSWORD: ${{ secrets.DOCKER_ACCESS_TOKEN }}
        run: |
          docker login -u ${{ secrets.DOCKER_USER }} -p $DOCKER_PASSWORD
      - name: Build Docker image
        run: docker build . --tag ${{ secrets.DOCKER_USER }}/reshape:${{ inputs.version }} --tag ${{ secrets.DOCKER_USER }}/reshape:latest
      - name: Push Docker image
        run: docker push --all-tags ${{ secrets.DOCKER_USER }}/reshape



================================================
FILE: .github/workflows/release.yaml
================================================
name: Release

on:
  release:
    types: [ created ]

env:
  CARGO_TERM_COLOR: always

jobs:
  build-binary:
    strategy:
      matrix:
        include:
        - name: Linux 64-bit
          file-name: linux_amd64
          target: x86_64-unknown-linux-gnu
          host: ubuntu-latest
          use-cross: true
        - name: Linux 32-bit
          file-name: linux_386
          target: i686-unknown-linux-gnu
          host: ubuntu-latest
          use-cross: true
        - name: Linux ARM 64-bit
          file-name: linux_aarch64
          target: aarch64-unknown-linux-gnu
          host: ubuntu-latest
          use-cross: true
        - name: macOS
          file-name: darwin_amd64
          target: x86_64-apple-darwin
          host: macos-11
          use-cross: false
        - name: macOS Apple Silicon
          file-name: darwin_aarch64
          target: aarch64-apple-darwin
          host: macos-11
          use-cross: false

    runs-on: ${{ matrix.host }}

    steps:
    - name: Checkout
      uses: actions/checkout@v2
    - name: Select Rust toolchain 
      uses: actions-rs/toolchain@v1
      with:
          toolchain: stable
          override: true
          target: ${{ matrix.target }}
    - name: Use cache for Rust dependencies
      uses: Swatinem/rust-cache@v1
    - name: Build
      uses: actions-rs/cargo@v1
      with:
        command: build
        args: --release --target ${{ matrix.target }}
        use-cross: ${{ matrix.use-cross }}
    - name: Rename binary
      run: mv target/${{ matrix.target }}/release/reshape ./reshape-${{ matrix.file-name }}
    - name: Upload binary to release
      env:
        GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
      run: gh release upload ${GITHUB_REF##*/} "reshape-${{ matrix.file-name }}#${{ matrix.name }}" --clobber

  publish-docker:
    runs-on: ubuntu-latest
    steps:
    - name: Checkout
      uses: actions/checkout@v2
    - name: Log in to Docker
      env:
        DOCKER_PASSWORD: ${{ secrets.DOCKER_ACCESS_TOKEN }}
      run: |
        docker login -u ${{ secrets.DOCKER_USER }} -p $DOCKER_PASSWORD
    - name: Build Docker image
      # GITHUB_REF is formatted as: refs/tags/v0.0.1
      # The shell expansion used below will remove everything up to the version number, leaving 0.0.1
      run: docker build . --tag ${{ secrets.DOCKER_USER }}/reshape:${GITHUB_REF##*/v} --tag ${{ secrets.DOCKER_USER }}/reshape:latest
    - name: Push Docker image
      run: docker push --all-tags ${{ secrets.DOCKER_USER }}/reshape

  publish-crate:
    runs-on: ubuntu-latest
    needs: ["build-binary", "publish-docker"]
    steps:
    - name: Checkout
      uses: actions/checkout@v2
    - name: Select Rust toolchain 
      uses: actions-rs/toolchain@v1
      with:
          toolchain: stable
    - name: Publish to crates.io
      uses: actions-rs/cargo@v1
      with:
        command: publish
      env:
        CARGO_REGISTRY_TOKEN: ${{ secrets.CRATES_IO_TOKEN }}


================================================
FILE: .github/workflows/test.yaml
================================================
name: Tests

on:
  push:
    branches: [ main ]
  pull_request:
    branches:
    - main

env:
  CARGO_TERM_COLOR: always

jobs:
  integration-tests:
    runs-on: ubuntu-latest

    services:
      postgres:
        image: postgres
        ports:
          - 5432:5432
        env:
          POSTGRES_DB: migra_test
          POSTGRES_USER: postgres
          POSTGRES_PASSWORD: postgres
        # Set health checks to wait until postgres has started
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5

    steps:
    - name: Checkout
      uses: actions/checkout@v2
    - name: Select Rust toolchain 
      uses: actions-rs/toolchain@v1
      with:
          toolchain: stable
    - name: Use cache for Rust dependencies
      uses: Swatinem/rust-cache@v1
    - name: Run integration tests
      uses: actions-rs/cargo@v1
      with:
        command: test
        args: -- --test-threads=1
      env:
        POSTGRES_CONNECTION_STRING: "postgres://postgres:postgres@127.0.0.1/migra_test"