clean_couch/main.go

package main

import (
	"encoding/json"
	"fmt"
	"net/http"
	"time"

	"github.com/cenkalti/backoff"
	"github.com/spf13/pflag"
)

var config = struct {
	CouchBaseURL string
	Database     string
	View         string
	Routines     int

	// Private storage
	totalNumberOfDocuments int
	processedDocuments     int
	processChannel         chan bool
	concurrencyChannel     chan bool
}{
	processChannel: make(chan bool, 10),
}

func main() {
	pflag.StringVar(&config.CouchBaseURL, "baseurl", "http://localhost:5984", "BaseURL of your CouchDB instance")
	pflag.StringVar(&config.Database, "database", "", "The database containing your view and the data to delete")
	pflag.StringVar(&config.View, "view", "", "The view selecting the data to delete")
	pflag.IntVar(&config.Routines, "concurrency", 20, "How many delete requests should get processed concurrently?")
	pflag.Parse()

	if config.Database == "" || config.View == "" {
		pflag.Usage()
		return
	}

	delData := struct {
		Rows []struct {
			ID  string `json:"id"`
			Rev string `json:"key"`
		} `json:"rows"`
	}{}

	err := backoff.Retry(func() error {
		req, _ := http.NewRequest("GET", fmt.Sprintf("%s/%s/%s", config.CouchBaseURL, config.Database, config.View), nil)
		res, err := http.DefaultClient.Do(req)
		if err != nil {
			return err
		}
		defer res.Body.Close()

		if err := json.NewDecoder(res.Body).Decode(&delData); err != nil {
			return err
		}

		return nil
	}, backoff.NewExponentialBackOff())
	if err != nil {
		fmt.Printf("Tried to get the view but did not succeed: %s", err)
		return
	}

	config.totalNumberOfDocuments = len(delData.Rows)
	config.processedDocuments = 0

	config.concurrencyChannel = make(chan bool, config.Routines)

	go func() {
		for _, row := range delData.Rows {
			// Blocks when concurrency channel is full
			config.concurrencyChannel <- true

			go func(finChan chan bool, conChan chan bool, id, rev string) {
				// Retry deletes
				bo := backoff.NewExponentialBackOff()
				bo.InitialInterval = 5 * time.Second

				err := backoff.Retry(func() error {
					url := fmt.Sprintf("%s/%s/%s?rev=%s", config.CouchBaseURL, config.Database, id, rev)
					req, _ := http.NewRequest("DELETE", url, nil)
					res, err := http.DefaultClient.Do(req)
					if err != nil {
						return err
					}
					res.Body.Close()
					return nil
				}, bo)
				if err != nil {
					fmt.Printf("Unable to delete document with ID %s", id)
				}
				// Increase finished counter
				finChan <- true

				// Remove self from concurrency limit
				<-conChan
			}(config.processChannel, config.concurrencyChannel, row.ID, row.Rev)
		}
	}()

	ticker := time.NewTicker(time.Second)

	for {
		select {
		case <-config.processChannel:
			config.processedDocuments++
			if config.processedDocuments == config.totalNumberOfDocuments {
				fmt.Print("\n\n")
				return
			}
		case <-ticker.C:
			fmt.Printf("Processed %d of %d documents.\r", config.processedDocuments, config.totalNumberOfDocuments)
		}
	}

}