day.go

package main

import (
	"database/sql"
	"fmt"
	"log"
	"strings"

	_ "github.com/go-sql-driver/mysql"
)

// ours and those without their own stimulated (with tag 72)
var a = `
select distinct(cr.contact_id_a) contact_id_a, cet.tag_id is not null
from civicrm_relationship cr
         inner join civicrm_contact cc on cr.contact_id_b = cc.id and cc.job_title is not null and cc.is_deleted = 0
         left outer join civicrm_entity_tag cet
                         on cr.contact_id_a = cet.entity_id and cet.entity_table = 'civicrm_contact' and cet.tag_id = 74
where cr.relationship_type_id = 19
union
distinct
select cc.id contact_id_a, cet2.tag_id is not null
from civicrm_contact cc
         inner join civicrm_entity_tag cet
                    on cc.id = cet.entity_id and cet.entity_table = 'civicrm_contact' and cet.tag_id = 72
         left outer join civicrm_entity_tag cet2
                         on cc.id = cet2.entity_id and cet2.entity_table = 'civicrm_contact' and cet2.tag_id = 74
order by contact_id_a
;`

// for each A, count the number of Bs from group x where no action was taken yet (job_title is null)
var aPendingInGroup = `
select cr.contact_id_a,
       SUM(CASE WHEN cet.tag_id is null THEN 1 ELSE 0 END),
       SUM(CASE WHEN cet.tag_id is null THEN 0 ELSE 1 END)
from civicrm_relationship cr
         inner join civicrm_contact cc on cr.contact_id_b = cc.id and cc.job_title is null and cc.is_deleted = 0
         inner join civicrm_group_contact gc
                    on cr.contact_id_b = gc.contact_id and gc.group_id = %d and gc.status = 'Added'
         left outer join civicrm_entity_tag cet
                         on cr.contact_id_b = cet.entity_id and cet.entity_table = 'civicrm_contact' and cet.tag_id = 74
where cr.relationship_type_id = 19
group by cr.contact_id_a
order by cr.contact_id_a;
`

// theirs from group x, not associated with any A
// excluding tag_id 71 - SkipDist
var b = `
select cc.id, cet2.tag_id is not null
from civicrm_group_contact gc
         inner join civicrm_contact cc on cc.id = gc.contact_id
         left outer join civicrm_relationship rr on rr.contact_id_b = gc.contact_id and rr.relationship_type_id = 19
         left outer join civicrm_entity_tag cet
                         on cc.id = cet.entity_id and cet.entity_table = 'civicrm_contact' and cet.tag_id = 71
         left outer join civicrm_entity_tag cet2
                         on cc.id = cet2.entity_id and cet2.entity_table = 'civicrm_contact' and cet2.tag_id = 74
where gc.group_id = %d
  and gc.status = 'Added'
  and cc.job_title IS NULL
  and cc.is_deleted = 0
  and rr.contact_id_a is NULL
  and cet.tag_id is null
order by cc.id
`

var c = "insert into civicrm_relationship (contact_id_a, contact_id_b, relationship_type_id, is_active) values %s"
var c1 = "(%d, %d, 19, 1)"

type Person struct {
	ID    int
	IsRus bool
}

type PersonCount struct {
	Rus   int
	NoRus int
}

func main() {
	db, err := sql.Open("mysql", "day-prod:ZV1Bp-UZv85MD_Jw@tcp(am11.org:6033)/day-prod")
	if err != nil {
		log.Fatal(err)
	}
	defer db.Close()

	fmt.Println("Connected successfully")

	aIDs := loadIDs(db, a, "a")
	log.Printf("Got %d As\n", len(aIDs))

	aPending264Hist := loadHistogram(db, fmt.Sprintf(aPendingInGroup, 264), "aPendingInGroup[264]")
	log.Printf("Got %d keys in aPending264Hist\n", len(aPending264Hist))

	b264IDs := loadIDs(db, fmt.Sprintf(b, 264), "b[264]")
	log.Printf("Got %d Bs\n", len(b264IDs))

	dist := distribute(aIDs, b264IDs, aPending264Hist)

	for k, v := range dist {
		values := make([]string, len(v))
		for i := range v {
			values[i] = fmt.Sprintf(c1, k, v[i])
		}
		query := fmt.Sprintf(c, strings.Join(values, ","))

		//fmt.Println(query)

		tx, err := db.Begin()
		if err != nil {
			log.Fatal("ERROR: Unable to begin DB tx", err)
		}

		if _, err = tx.Exec(query); err != nil {
			fmt.Printf("ERROR: DB insert error: %+v\n", err)
			tx.Rollback()
			break
		}

		if err := tx.Commit(); err != nil {
			fmt.Printf("ERROR: Unable to commit DB tx: %+v\n", err)
		}
	}
}

func distribute(aIDs []*Person, bIDs []*Person, ahist map[int]*PersonCount) map[int][]int {
	dist := make(map[int][]int, len(aIDs))

	bNotRus := make([]*Person, 0)
	bRus := make([]*Person, 0)
	for _, x := range bIDs {
		if x.IsRus {
			bRus = append(bRus, x)
		} else {
			bNotRus = append(bNotRus, x)
		}
	}
	fmt.Printf("B's norus %d, rus %d\n", len(bNotRus), len(bRus))

	pageSize := 10

	// flat, equal, blind, distribution
	//for i := range aIDs {
	//	pos := i * pageSize
	//	dist[aIDs[i]] = bIDs[pos : pos+pageSize]
	//}

	// capacity aware distribution
	bRusIdx := 0
	bNoRusIdx := 0
	var bIdx *int
	for i := range aIDs {
		bsPool := bNotRus
		bIdx = &bNoRusIdx

		relCount := 0
		pc, ok := ahist[aIDs[i].ID]
		if ok {
			relCount = pc.NoRus
			if aIDs[i].IsRus {
				relCount = pc.Rus
			}
		}

		bFromRus := false
		if aIDs[i].IsRus && bRusIdx < len(bRus) {
			bsPool = bRus
			bIdx = &bRusIdx
			bFromRus = true
		}

		bs := pageSize - relCount
		if bs <= 0 {
			continue
		}

		// boundary checks
		count := Min(len(bsPool)-*bIdx, bs)
		if count < 1 {
			fmt.Printf("Not enough Bs for A[%d][%t] Bs[%t] bRusIdx %d bNoRusIdx %d\n", aIDs[i].ID, aIDs[i].IsRus, bFromRus, bRusIdx, bNoRusIdx)
			continue
		}

		ids := make([]int, count)
		for j := 0; j < count; j++ {
			ids[j] = bsPool[*bIdx+j].ID
		}
		dist[aIDs[i].ID] = ids
		fmt.Printf("distributing %d Bs[%t] to A[%d][%t]\n", len(ids), bFromRus, aIDs[i].ID, aIDs[i].IsRus)

		*bIdx += count
	}

	fmt.Printf("%d As need more Bs\n", len(dist))
	return dist
}

func loadIDs(db *sql.DB, query string, name string) []*Person {
	rows, err := db.Query(query)
	if err != nil {
		log.Fatal(name, "db.Query", err)
	}

	ids := make([]*Person, 0)
	for rows.Next() {
		var id int
		var isRus bool
		if err := rows.Scan(&id, &isRus); err != nil {
			log.Fatal(name, "rows.Scan id", err)
		}
		ids = append(ids, &Person{ID: id, IsRus: isRus})
	}

	if err := rows.Err(); err != nil {
		log.Fatal(name, "rows.Err", err)
	}

	return ids
}

func loadHistogram(db *sql.DB, query string, name string) map[int]*PersonCount {
	rows, err := db.Query(query)
	if err != nil {
		log.Fatal(name, "db.Query", err)
	}

	hist := make(map[int]*PersonCount, 0)
	for rows.Next() {
		var k, noRusCount, rusCount int
		if err := rows.Scan(&k, &noRusCount, &rusCount); err != nil {
			log.Fatal(name, "rows.Scan k,norus, rus", err)
		}
		hist[k] = &PersonCount{
			NoRus: noRusCount,
			Rus:   rusCount,
		}
	}

	if err := rows.Err(); err != nil {
		log.Fatal(name, "rows.Err", err)
	}

	return hist
}

// Like math.Min for int
func Min(x, y int) int {
	if x < y {
		return x
	}
	return y
}

remove_russian_associations.sql

delete
from civicrm_relationship
where id in (
    select *
    from (select cr.id
          from civicrm_relationship cr
                   inner join civicrm_entity_tag cet
                              on cr.contact_id_a = cet.entity_id and cet.entity_table = 'civicrm_contact' and
                                 cet.tag_id = 74
                   inner join civicrm_group_contact gc
                              on cr.contact_id_b = gc.contact_id and gc.group_id = 264 and gc.status = 'Added'
                   inner join civicrm_contact cc on cr.contact_id_b = cc.id and cc.job_title is null
          where cr.relationship_type_id = 19) as tmp);

delete
from civicrm_relationship
where contact_id_a = 4689
  and contact_id_b in (
    select *
    from (select distinct cc.id
          from civicrm_contact cc
                   left join civicrm_group_contact gc on cc.id = gc.contact_id
                   left join civicrm_entity_tag cet on cet.entity_id = cc.id
          where cc.id in (select distinct contact_id_b
                          from civicrm_relationship
                          where contact_id_a = 4689
                            and relationship_type_id = 19)
            and cc.job_title is null
          order by cc.id) as tmp);



select cc.id,
       cc.sort_name,
       SUM(CASE WHEN cet2.tag_id is null THEN 1 ELSE 0 END) norus,
       SUM(CASE WHEN cet2.tag_id is null THEN 0 ELSE 1 END) rus
from civicrm_relationship cr
         inner join civicrm_contact cc on cr.contact_id_a = cc.id
         inner join civicrm_entity_tag cet on cr.contact_id_a = cet.entity_id and cet.tag_id = 74
         inner join civicrm_group_contact gc
                    on cr.contact_id_b = gc.contact_id and gc.group_id = 264 and gc.status = 'Added'
         inner join civicrm_contact cc2 on cr.contact_id_b = cc2.id and cc2.job_title is null
         left join civicrm_entity_tag cet2 on cr.contact_id_b = cet2.entity_id and cet2.tag_id = 74
group by cr.contact_id_a
having count(cr.id) > 10
order by count(cr.id) desc;

select *
from civicrm_entity_tag
where entity_id in
      (
          select cc.id, cc.sort_name, count(cr.id)
          from civicrm_relationship cr
                   inner join civicrm_contact cc on cr.contact_id_a = cc.id
                   inner join civicrm_entity_tag cet on cr.contact_id_a = cet.entity_id and cet.tag_id = 74
                   inner join civicrm_entity_tag cet2 on cr.contact_id_b = cet2.entity_id and cet2.tag_id = 74
                   inner join civicrm_group_contact gc on cr.contact_id_b = gc.contact_id and gc.group_id = 264
          group by cr.contact_id_a
          having count(cr.id) > 10
      )
order by entity_id;



select cc.id,
       cc.sort_name,
       count(cr.id)
from civicrm_relationship cr
         inner join civicrm_contact cc on cr.contact_id_a = cc.id
         inner join civicrm_entity_tag cet on cr.contact_id_a = cet.entity_id and cet.tag_id = 74
         inner join civicrm_group_contact gc
                    on cr.contact_id_b = gc.contact_id and gc.group_id = 264 and gc.status = 'Added'
         inner join civicrm_contact cc2 on cr.contact_id_b = cc2.id and cc2.job_title is null
#          left join civicrm_entity_tag cet2 on cr.contact_id_b = cet2.entity_id and cet2.tag_id = 74
group by cr.contact_id_a
having count(cr.id) > 10
order by count(cr.id) desc;




select cr.contact_id_a,
       SUM(CASE WHEN cet.tag_id is null THEN 1 ELSE 0 END),
       SUM(CASE WHEN cet.tag_id is null THEN 0 ELSE 1 END)
from civicrm_relationship cr
         inner join civicrm_contact cc on cr.contact_id_b = cc.id and cc.job_title is null and cc.is_deleted = 0
         inner join civicrm_group_contact gc
                    on cr.contact_id_b = gc.contact_id and gc.group_id = 264 and gc.status = 'Added'
         left outer join civicrm_entity_tag cet
                         on cr.contact_id_b = cet.entity_id and cet.entity_table = 'civicrm_contact' and cet.tag_id = 74
where cr.relationship_type_id = 19
group by cr.contact_id_a
order by cr.contact_id_a;



delete
from civicrm_relationship
where id in (
    select *
    from (select distinct cr.id from civicrm_entity_tag cet
 inner join civicrm_relationship cr on cet.entity_id = cr.contact_id_b and cr.relationship_type_id = 19
 inner join civicrm_contact cc on cr.contact_id_b = cc.id and cc.job_title is null and cc.is_deleted = 0
 inner join civicrm_group_contact gc on cr.contact_id_b = gc.contact_id and gc.group_id = 264 and gc.status = 'Added'
where cet.tag_id = 74) as tmp);



delete
from civicrm_relationship
where id in (
    select *
    from (select distinct cr.id from civicrm_relationship cr
                                         inner join civicrm_contact cc on cr.contact_id_b = cc.id and cc.job_title is null and cc.is_deleted = 0
                                         inner join civicrm_group_contact gc on cr.contact_id_b = gc.contact_id and gc.group_id = 264 and gc.status = 'Added'
          where cr.relationship_type_id = 19) as tmp);